Correct issue where StartStream callback could be invoked multiple ti…

…mes.

Change-Id: I9012973a2cf50366ec8f2f846085c0ac5880edfe
Reviewed-on: http://review.couchbase.org/63827
Reviewed-by: Michael Wiederhold <mike@couchbase.com>
Tested-by: Brett Lawson <brett19@gmail.com>

gocbcore/agentops_dcp.go

@@ -41,14 +41,13 @@ type GetVBucketSeqnosCallback func(uint16, SeqNo, error)
 func (c *Agent) OpenStream(vbId uint16, vbUuid VbUuid, startSeqNo, endSeqNo, snapStartSeqNo, snapEndSeqNo SeqNo, evtHandler StreamObserver, cb OpenStreamCallback) (PendingOp, error) {
 	var req *memdQRequest
 	handler := func(resp *memdResponse, err error) {
-		if err != nil {
-			// All client errors are handled by the StreamObserver
-			cb(nil, err)
-			return
-		}
-
 		if resp.Magic == ResMagic {
 			// This is the response to the open stream request.
+			if err != nil {
+				// All client errors are handled by the StreamObserver
+				cb(nil, err)
+				return
+			}
 
 			numEntries := len(resp.Value) / 16
 			entries := make([]FailoverEntry, numEntries)
@@ -63,6 +62,11 @@ func (c *Agent) OpenStream(vbId uint16, vbUuid VbUuid, startSeqNo, endSeqNo, sna
 			return
 		}
 
+		if err != nil {
+			evtHandler.End(vbId, err)
+			return
+		}
+
 		// This is one of the stream events
 		switch resp.Opcode {
 		case CmdDcpSnapshotMarker: