Merge branch 'pr/80'

Change-Id: If42c34099c45b1a5932fb0471e17ba6624694cbd

Author: skriptble

core/connection/connection.go

@@ -358,7 +358,7 @@ func canMonitor(cmd string) bool {
 	return true
 }
 
-func (c *connection) commandStartedEvent(wm wiremessage.WireMessage) error {
+func (c *connection) commandStartedEvent(ctx context.Context, wm wiremessage.WireMessage) error {
 	if c.cmdMonitor == nil || c.cmdMonitor.Started == nil {
 		return nil
 	}
@@ -417,7 +417,7 @@ func (c *connection) commandStartedEvent(wm wiremessage.WireMessage) error {
 		startedEvent.Command = emptyDoc
 	}
 
-	c.cmdMonitor.Started(startedEvent)
+	c.cmdMonitor.Started(ctx, startedEvent)
 
 	if !acknowledged {
 		if c.cmdMonitor.Succeeded == nil {
@@ -432,7 +432,7 @@ func (c *connection) commandStartedEvent(wm wiremessage.WireMessage) error {
 			ConnectionID:  c.id,
 		}
 
-		c.cmdMonitor.Succeeded(&event.CommandSucceededEvent{
+		c.cmdMonitor.Succeeded(ctx, &event.CommandSucceededEvent{
 			CommandFinishedEvent: finishedEvent,
 			Reply: bson.NewDocument(
 				bson.EC.Int32("ok", 1),
@@ -489,7 +489,7 @@ func processReply(reply *bson.Document) (bool, string) {
 	return false, fullErrMsg
 }
 
-func (c *connection) commandFinishedEvent(wm wiremessage.WireMessage) error {
+func (c *connection) commandFinishedEvent(ctx context.Context, wm wiremessage.WireMessage) error {
 	if c.cmdMonitor == nil {
 		return nil
 	}
@@ -532,7 +532,7 @@ func (c *connection) commandFinishedEvent(wm wiremessage.WireMessage) error {
 				Reply:                emptyDoc,
 				CommandFinishedEvent: finishedEvent,
 			}
-			c.cmdMonitor.Succeeded(successEvent)
+			c.cmdMonitor.Succeeded(ctx, successEvent)
 			return nil
 		}
 
@@ -553,7 +553,7 @@ func (c *connection) commandFinishedEvent(wm wiremessage.WireMessage) error {
 			CommandFinishedEvent: finishedEvent,
 		}
 
-		c.cmdMonitor.Succeeded(successEvent)
+		c.cmdMonitor.Succeeded(ctx, successEvent)
 		return nil
 	}
 
@@ -562,7 +562,7 @@ func (c *connection) commandFinishedEvent(wm wiremessage.WireMessage) error {
 		CommandFinishedEvent: finishedEvent,
 	}
 
-	c.cmdMonitor.Failed(failureEvent)
+	c.cmdMonitor.Failed(ctx, failureEvent)
 	return nil
 }
 
@@ -639,7 +639,7 @@ func (c *connection) WriteWireMessage(ctx context.Context, wm wiremessage.WireMe
 	}
 
 	c.bumpIdleDeadline()
-	err = c.commandStartedEvent(wm)
+	err = c.commandStartedEvent(ctx, wm)
 	if err != nil {
 		return err
 	}
@@ -790,7 +790,7 @@ func (c *connection) ReadWireMessage(ctx context.Context) (wiremessage.WireMessa
 	}
 
 	c.bumpIdleDeadline()
-	err = c.commandFinishedEvent(wm)
+	err = c.commandFinishedEvent(ctx, wm)
 	if err != nil {
 		return nil, err // TODO: do we care if monitoring fails?
 	}

core/event/monitoring.go

@@ -7,6 +7,7 @@
 package event
 
 import (
+	"context"
 	"time"
 
 	"github.com/mongodb/mongo-go-driver/bson"
@@ -64,7 +65,7 @@ type CommandFailedEvent struct {
 
 // CommandMonitor represents a monitor that is triggered for different events.
 type CommandMonitor struct {
-	Started   func(*CommandStartedEvent)
-	Succeeded func(*CommandSucceededEvent)
-	Failed    func(*CommandFailedEvent)
+	Started   func(context.Context, *CommandStartedEvent)
+	Succeeded func(context.Context, *CommandSucceededEvent)
+	Failed    func(context.Context, *CommandFailedEvent)
 }

mongo/causal_consistency_test.go

@@ -7,6 +7,7 @@
 package mongo
 
 import (
+	"context"
 	"os"
 	"reflect"
 	"testing"
@@ -23,10 +24,10 @@ var ccStarted *event.CommandStartedEvent
 var ccSucceeded *event.CommandSucceededEvent
 
 var ccMonitor = &event.CommandMonitor{
-	Started: func(cse *event.CommandStartedEvent) {
+	Started: func(ctx context.Context, cse *event.CommandStartedEvent) {
 		ccStarted = cse
 	},
-	Succeeded: func(cse *event.CommandSucceededEvent) {
+	Succeeded: func(ctx context.Context, cse *event.CommandSucceededEvent) {
 		ccSucceeded = cse
 	},
 }

mongo/command_monitoring_test.go

@@ -38,13 +38,13 @@ var failedChan = make(chan *event.CommandFailedEvent, 100)
 var cursorID int64
 
 var monitor = &event.CommandMonitor{
-	Started: func(cse *event.CommandStartedEvent) {
+	Started: func(ctx context.Context, cse *event.CommandStartedEvent) {
 		startedChan <- cse
 	},
-	Succeeded: func(cse *event.CommandSucceededEvent) {
+	Succeeded: func(ctx context.Context, cse *event.CommandSucceededEvent) {
 		succeededChan <- cse
 	},
-	Failed: func(cfe *event.CommandFailedEvent) {
+	Failed: func(ctx context.Context, cfe *event.CommandFailedEvent) {
 		failedChan <- cfe
 	},
 }

mongo/sessions_test.go

@@ -53,10 +53,10 @@ var sessionSucceeded *event.CommandSucceededEvent
 var sessionsMonitoredTop *topology.Topology
 
 var sessionsMonitor = &event.CommandMonitor{
-	Started: func(cse *event.CommandStartedEvent) {
+	Started: func(ctx context.Context, cse *event.CommandStartedEvent) {
 		sessionStarted = cse
 	},
-	Succeeded: func(cse *event.CommandSucceededEvent) {
+	Succeeded: func(ctx context.Context, cse *event.CommandSucceededEvent) {
 		sessionSucceeded = cse
 	},
 }