js: Add js.PullSubscribe and sub.Fetch APIs for pull consumers

[wallyqs]

This changes the way how Pull consumers work by adding the two following APIs:

func (js *js) PullSubscribe(subj string, opts ...SubOpt) (*Subscription, error)
func (sub *Subscription) Fetch(batch int, opts ...PullOpt) ([]*Msg, error)

PullSubscribe replaces using SubscribeSync or QueueSubscribeSync along with nats.Pull option to create pull based consumers. When a subscription is called with PullSubscribe, it can use sub.Fetch() to get a synchronous response of a collection of messages.

Example usage

js, _ := nc.JetStream()

// Create pull based consumer with 128 max concurrent inflight pull requests.
sub, err := js.PullSubscribe("foo", nats.Durable("example"), nats.PullMaxWaiting(128))

// Simple loop to fetch one message is efficient:
// internally first makes a request to get a single message if
// there is one already, if none then makes longer request
// that waits up to 2 seconds
for {
        msgs, err := sub.Fetch(1, nats.MaxWait(2*time.Second))
        if err != nil {
                log.Println("Error during fetch:", err)
		continue
        }

	// If no errors, it is ensured that there would always be at
	// least a message at this point.
	msg := msgs[0]
	msg.Ack()
}

// When expecting for multiple messages, it will be awaited
// for the delivery of all the messages or yield the delivered
// results when the timeout expires.
for {
        msgs, err := sub.Fetch(10, nats.MaxWait(2*time.Second))
        if err != nil {
                log.Println("Error during fetch:", err)
		continue
        }
	for _, msg := range msgs {
		msg.Ack()
	}
}

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

[coveralls]

Coverage decreased (-0.4%) to 86.657% when pulling 99d20d3 on js-pull-subscribe into d70f82c on master.

[derekcollison]

Let's squash down for review.

[wallyqs]

Squashed and updated server version in go.mod

[scottf]

What is this?

// Deletes durable consumer on unsubscribe
defer sub.Unsubscribe()

[wallyqs]

@scottf that means that when Unsubscribe() is called then also makes the consumer go away:

https://github.com/nats-io/nats.go/blob/master/js.go#L427-L440
https://github.com/nats-io/nats.go/blob/master/nats.go#L3567-L3576'

Note: This comment is stale as the original example has changed.

[wallyqs]

Rebased and updated to test against latest version of the server.

[wallyqs]

In case only Batch is set, could take advantage of the fast path and just the number as bytes.

[ColinSullivan1]

Also results in a lighter protocol.

[ColinSullivan1]

Should there be another error type surfaced that might be more meaningful?

[ColinSullivan1]

Convert this case for > 400 to future proof this? (either convert to int or test the first char to '4' or '5')... Alternatively, could add a default that results in an error. wdyt?

[ColinSullivan1]

LGTM; the suggested changes aren't critical to this PR.

[wallyqs]

This is rebased and updated against latest version of the server. The logic is now smarter using NoWait to be signaled in case there are no messages and fallback to a longer pull batch request in case there aren't any at the moment and wait for the delivery. In case of batches of 1, also the way pull works was simplified to be a mix of new style request plus an old style request for the second one in case the initial NoWait request fails with no messages.

[derekcollison]

In general LGTM