changeStream support

[peterdeka]

Already merged the development branch into this.

The changeStream now makes use of the MaxAwaitTimeMS option by setting the timeout property of the underlying iterator (opinions welcome, don-t really know if it should be a tailable instead) also on the 0 value handling). From this follows that Next() can now time out and no more block forever.

Error handling was refactored a bit in order to handle the timeout, taking inspiration from other calls that can timeout i added a Timeout() function. When Next() returns false, one should decide if there was a timeout or a bad error based on the calls to Timeout() and Err().
As a matter fo fact a ErrNotFound coming from the underlying iterator is not processed through the resume cycle as it is not related to socket/connection errors and that would imply a great waste of resources (as we expect Next() to be called in some sort of cycles).

The resumable error policy was changed a bit in order to categorize a missing resume token as a non resumable error as it does not depend on the connection but on a non solvable problem (as the pipeline removing it is specified during Watch() and cannot be changed).
The latest resume token is now taken in account (was not used before) and the resume/session killing process was refactored in order to solve leaking sockets.

Test were written, have no idea at the moment on how to test the resume cycle (kill something?), test run was all green on Travis. I had to edit the .env options script in order to remove the now deprecated --nojournal option that was failing all my streams tests (RS were not joining on 3.6.2 to be honest).

Review and real world testing would be very appreciated, regarding the real world i hope to deploy asap.

[peterdeka]

Hi @domodwyer yep i started from your feature/changestream branch, merged the current development and did my work. So it should merge right away back in.
Far too kind before review 😅

[peterdeka]

[UPDATE] i updated the PR, now we are using the MaxAwaitTimeMS option in a native way (the option is passed to the aggregation instead of being artificial in the driver's iterator Next() method). In order to achieve this i added the MaxTimeMS method to the Pipe and cascaded the option down to the iterator as in the case of the ChangeStream it's a getMore option. Moved to a TODO the Collation parameter in the stream options as it's useless at the moment, connected the batch size option. Still have to find a way to test the resume function.

[domodwyer]

Hey @peterdeka

Can you rebase this all onto the development branch and we'll just go for a direct merge into it?

The diffs are messed up (I had to cut a new branch to try fix it) and it's looks like there's not going to be any reason not to get it into development.

Thanks for the hard work!
Dom

[domodwyer]

Feel free to open a new PR against development if it's easier or keep this one - whatever is best for you 👍

Dom

[domodwyer]

It's probably nicer to expose this as MaxTime(d time.Duration) and convert to milliseconds internally.

[domodwyer]

Or QueryTimeout()? SetQueryTimeout()?

[peterdeka]

Absolutely right, i'm changing the parameter to a time.Duration. Regarding the naming, i chose the same name as the parameter just to be no brainer and map it 1 to 1 but it is definetly not a problem, you decide! Tell me what you think it's the best to be consistent with the rest of the codebase.

[domodwyer]

We already have cluster.SetSyncTimeout() so maybe SetQueryTimeout() - what do you think?

[peterdeka]

As we already have func (q *Query) SetMaxTime(d time.Duration) i would stick with a SetMaxTime just to avoid citing query in the name.

[domodwyer]

Sounds good to me!

[peterdeka]

@domodwyer should be ok now

[domodwyer]

Hi @peterdeka

First off: that's some truly lovely code to read! I wish the rest of the mgo code base was as clear and readable as this PR. I've left one comment regarding passing in a time.Duration instead of a uint of milliseconds, but after that it's definitely good to merge.

Great work and thanks, it's really appreciated!

Dom

[domodwyer]

@peterdeka thanks very much!

We shall get this tested and merged to master as soon as possible 👍

Thanks for the hard work - no doubt there will be a lot of people who use $changeStream who will be very thankful!

Dom

[piavgh]

Hi @peterdeka

I got the error cursor id 7412681357123903724 is already in use after 3 or 4 changes in my collection. Do you know which piece of code causes this issue? And maybe how to fix this issue?

This is my function to handle change streams

func (s *OrderService) handleChangeStream(ctx context.Context, ct *mgo.ChangeStream) {
	for {
		select {
		case <-ctx.Done(): // if parent context was cancelled
			err := ct.Close() // close the stream
			if err != nil {
				logger.Error("Change stream closed")
			}
			return //exiting from the func
		default:
			ev := types.OrderChangeEvent{}

			//getting next item from the steam
			ok := ct.Next(&ev)

			//if data from the stream wasn't un-marshaled, we get ok == false as a result
			//so we need to call Err() method to get info why
			//it'll be nil if we just have no data
			if !ok {
				err := ct.Err()
				if err != nil {
					//if err is not nil, it means something bad happened, let's finish our func
					logger.Error(err)
					return
				}
			}

			//if item from the stream un-marshaled successfully, do something with it
			if ok {
				logger.Debug(ev.OperationType)
				s.HandleDocumentType(ev)
			}
		}
	}
}

The error is logged from here

err := ct.Err()
if err != nil {
    //if err is not nil, it means something bad happened, let's finish our func
    logger.Error(err)
    return
}

Thank you.