js: FlowControl + Heartbeats for Push Consumers

[wallyqs]

This adds the following options to push based consumers:

sub, err := js.Subscribe("foo", func(msg *nats.Msg) {}, nats.EnableFlowControl(), nats.IdleHeartbeat(2*time.Second))

sub, err := js.SubscribeSync("foo", nats.EnableFlowControl(), nats.IdleHeartbeat(2*time.Second))

mch := make(chan *Msg, 512)
sub, err := js.ChanSubscribe("foo", mch, nats.EnableFlowControl(), nats.IdleHeartbeat(2*time.Second))

When they are enabled, the client will automatically handle/skip the control messages sent by the server. In case of push based consumers with Heartbeats, an asynchronous typed error will be dispatched so that the subscription could deleted and restarted from the given sequence:

errHandler := nats.ErrorHandler(func(c *nats.Conn, sub *nats.Subscription, err error) {
	if info, ok := err.(*nats.ErrConsumerSequenceMismatch); ok {
		t.Logf("Subscription should be restarted from stream sequence at %+v", info.StreamResumeSequence)
	}
})

Signed-off-by: Waldemar Quevedo wally@synadia.com

[wallyqs]

In order to support ChanSubscribe, SyncSubscribe and Async Subscribers, a goroutine that expects signals via a channel is started to process the status messages outside of the message delivery.

[derekcollison]

Why do we need this?

[wallyqs]

I moved the control messages handling to be done before delivering the message at processMsg and added an extra goroutine that would be handling the flow control messages from a subscription instead: https://github.com/nats-io/nats.go/pull/692/files#diff-1d2fb9af677137c0182213770500dd5625a51ba7286301cb11d0197b21e0f948R2533

[derekcollison]

Anyway to avoid having another Go routine?

[wallyqs]

I removed the goroutine and now responding a flow control message happens as soon as we receive the message, we have to do the processing of control messages there to be able to support ChanSubscribe:

• https://github.com/nats-io/nats.go/pull/692/files#diff-1d2fb9af677137c0182213770500dd5625a51ba7286301cb11d0197b21e0f948R2543
• https://github.com/nats-io/nats.go/pull/692/files#diff-176a2342c55218b70f165bed6a7731dc83d901f9513382f3c98bbd3d08ce0678R1047

[coveralls]

Coverage decreased (-0.3%) to 86.339% when pulling 1b16e34 on js-hbs-chan into a0b1f60 on master.

[derekcollison]

LGTM