If there was an error writing to the socket, this code should return from the function. Currently this loop will keep receiving events and trying to write them forever.

Shouldn't this operate on a copy of e, so that the event itself (which is shared between all the subscribers) is not modified by each subscriber?

@aaronlehmann yeah, I intended to do that, but must have forgotten. I think doing a deepcopy of the event is the way to go.

We shouldn't create the timer in the case where --until is not specified. It would fire immediately, and the select would always be able to take the <-timer.C branch, hogging CPU.

How about something like this:

var (
        timer *time.Timer
        timerCh <-chan time.Time
)
if until > 0 {
        dur := time.Unix(until, 0).Sub(time.Now())
        timer = time.NewTimer(dur)
        timerCh = timer.C
}

Then have a case <-timerCh: in the select below.

This is a better idea.

Don't need an else because of the return above.

Is eChan not an event?

eChan has an interface type, which we cast to *cluster.Event below.

Why not write directly here, rather than reallocate the data as a string? w.Write(data) should be sufficient.

Depending on go version, you can now use r.Context().Done() to listen to this cancellation.

I'm not sure the request context is set correctly in all cases in Swarm, some of the code is from before contexts were introduced in Go. I'll check.

This should be from net/http, so should be wired to the CloseNotifier.

@aaronlehmann Does this need "closed" detection? For example, will the watch ever shutdown this channel?

Currently it will not, but moby/swarmkit#2285 adds this as an option. So it probably makes sense to add the check here in anticipation of using that later.

Why don't we adapt this to the eventHandlers list? I am not sure that making this dependent on the watch queue makes sense.

@stevvooe I agree, but I also think that the eventual queue we will used will be moved upstream to go-events. This is supposed to be a temporary fix.

eventHandlers will go away actually, and all events should be handled using the watchQueue. For now though, some internal rescheduling functionality uses events which I did not want to mess with in this PR, that's why we have this case.

@nishanttotla That still doesn't explain why this isn't just a handler. You can do this with a closure to adapt Publish to Handler.

@stevvooe not sure I follow exactly here. What do you envision this handler to look like?

Not sure what you mean: just make an event handler. I am not sure how this got merged in its current state. It just doesn't make any sense to have both a handler set and a publish set. I'd recommend cleaning this up before it gets too far. It is going to be a pain to remove later and will confuse future contributors.

I will clean that up soon. The handler will entirely go away as it is, and only the publish set will remain.

Now that moby/swarmkit#2285 is merged, let's revendor swarmkit and change this constructor to:

NewQueue(WithTimeout(10 * time.Second), WithLimit(10000), WithCloseOutChan())

We probably want to define these values as config vars and plumb them through the standard config model used by classic swarm.
Assuming 1KB per event (gross overestimation), the size limit of 10000 for the queue should occupy no more than 10MB per listener in memory. For the edge case of 100 simultaneous inefficient listeners, this would lead up to to an additional 1 GB of memory

@alexmavr done. PTAL.

Should we consider event filtering here?

@alexmavr maybe we could do it as a follow up?