| 1 | package hostfiles |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "log" |
| 6 | "sync" |
| 7 | "time" |
| 8 | ) |
| 9 | |
| 10 | // PlatformStager is the part of a session's Publisher a PlatformRelay |
| 11 | // needs. *Publisher implements it. |
| 12 | type PlatformStager interface { |
| 13 | StagePlatform(ctx context.Context, name string, body []byte) error |
| 14 | } |
| 15 | |
| 16 | // PlatformRelay hands one platform item to a running session's publisher |
| 17 | // without making its caller wait on the store (instance-key plan ruling |
| 18 | // 22). Publish records the latest body and returns at once; Run stages it |
| 19 | // with StagePlatform, retrying until it is staged, and the publisher's own |
| 20 | // continuous loop (Run in run.go) commits it within its deadline, as it |
| 21 | // commits every staged change. A body staged once is not staged again |
| 22 | // unless Publish is called with it again. |
| 23 | type PlatformRelay struct { |
| 24 | Name string |
| 25 | Stage PlatformStager |
| 26 | Retry time.Duration // between failed stages; 0 is one second |
| 27 | |
| 28 | mu sync.Mutex |
| 29 | pending []byte |
| 30 | wake chan struct{} |
| 31 | } |
| 32 | |
| 33 | func (r *PlatformRelay) signal() chan struct{} { |
| 34 | r.mu.Lock() |
| 35 | defer r.mu.Unlock() |
| 36 | if r.wake == nil { |
| 37 | r.wake = make(chan struct{}, 1) |
| 38 | } |
| 39 | return r.wake |
| 40 | } |
| 41 | |
| 42 | // Publish queues body as the item's next value. It never blocks. |
| 43 | func (r *PlatformRelay) Publish(body []byte) { |
| 44 | w := r.signal() |
| 45 | r.mu.Lock() |
| 46 | r.pending = append([]byte(nil), body...) |
| 47 | r.mu.Unlock() |
| 48 | select { |
| 49 | case w <- struct{}{}: |
| 50 | default: |
| 51 | } |
| 52 | } |
| 53 | |
| 54 | // Run stages queued bodies until ctx ends, and returns ctx.Err(). |
| 55 | func (r *PlatformRelay) Run(ctx context.Context) error { |
| 56 | retry := r.Retry |
| 57 | if retry <= 0 { |
| 58 | retry = time.Second |
| 59 | } |
| 60 | w := r.signal() |
| 61 | failing := false // logged on the transitions only, as Run in run.go logs a scan |
| 62 | for { |
| 63 | r.mu.Lock() |
| 64 | body := r.pending |
| 65 | r.mu.Unlock() |
| 66 | if body != nil { |
| 67 | if err := r.Stage.StagePlatform(ctx, r.Name, body); err != nil { |
| 68 | if ctx.Err() != nil { |
| 69 | return ctx.Err() |
| 70 | } |
| 71 | if !failing { |
| 72 | failing = true |
| 73 | log.Printf("hostfiles: %s: stage failing, retrying every %s: %v", r.Name, retry, err) |
| 74 | } |
| 75 | select { |
| 76 | case <-ctx.Done(): |
| 77 | return ctx.Err() |
| 78 | case <-time.After(retry): |
| 79 | } |
| 80 | continue |
| 81 | } |
| 82 | if failing { |
| 83 | failing = false |
| 84 | log.Printf("hostfiles: %s: stage recovered", r.Name) |
| 85 | } |
| 86 | r.mu.Lock() |
| 87 | if string(r.pending) == string(body) { |
| 88 | r.pending = nil |
| 89 | } |
| 90 | r.mu.Unlock() |
| 91 | continue |
| 92 | } |
| 93 | select { |
| 94 | case <-ctx.Done(): |
| 95 | return ctx.Err() |
| 96 | case <-w: |
| 97 | } |
| 98 | } |
| 99 | } |
| 100 | |
amadan