WIP

Author: qingyang-hu

internal/integration/mtest/mongotest.go

@@ -364,7 +364,9 @@ func (t *T) ResetClient(opts *options.ClientOptions) {
 		t.clientOpts = opts
 	}
 
-	_ = t.Client.Disconnect(context.Background())
+	if t.Client != nil {
+		_ = t.Client.Disconnect(context.Background())
+	}
 	t.createTestClient()
 	t.DB = t.Client.Database(t.dbName)
 	t.Coll = t.DB.Collection(t.collName, t.collOpts)

internal/integration/sessions_test.go

@@ -13,6 +13,7 @@ import (
 	"fmt"
 	"reflect"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -597,46 +598,70 @@ func TestSessionsProse(t *testing.T) {
 		})
 	*/
 
+	mt.ResetClient(options.Client())
+	client := mt.Client
+	heartbeatStarted := make(chan struct{})
+	heartbeatSucceeded := make(chan struct{})
+	var clusterTimeAdvanced uint32
+	serverMonitor := &event.ServerMonitor{
+		ServerHeartbeatStarted: func(e *event.ServerHeartbeatStartedEvent) {
+			fmt.Println("Server heartbeat started:", e.ConnectionID)
+			if atomic.LoadUint32(&clusterTimeAdvanced) == 1 {
+				fmt.Println("ServerHeartbeatStartedEvent: cluster time advanced")
+				select {
+				case heartbeatStarted <- struct{}{}:
+					// NOOP
+				default:
+					// NOOP
+				}
+			}
+		},
+		ServerHeartbeatSucceeded: func(e *event.ServerHeartbeatSucceededEvent) {
+			fmt.Println("Server heartbeat succeeded:", e.ConnectionID, e.Duration, e.Reply)
+			if atomic.LoadUint32(&clusterTimeAdvanced) == 1 {
+				fmt.Println("ServerHeartbeatSucceededEvent: cluster time advanced")
+				select {
+				case heartbeatSucceeded <- struct{}{}:
+					// NOOP
+				default:
+					// NOOP
+				}
+			}
+		},
+	}
 	pingOpts := mtest.NewOptions().
 		CreateCollection(false).
-		Topologies(mtest.ReplicaSet, mtest.LoadBalanced)
+		ClientOptions(options.Client().
+			SetServerMonitor(serverMonitor).
+			SetHeartbeatInterval(500 * time.Millisecond). // Minimum interval
+			SetDirect(true)).
+		ClientType(mtest.Pinned)
 	mt.RunOpts("ping test", pingOpts, func(mt *mtest.T) {
-		serverMonitor := &event.ServerMonitor{
-			ServerHeartbeatStarted: func(e *event.ServerHeartbeatStartedEvent) {
-				fmt.Println("Server heartbeat started:", e.ConnectionID)
-			},
-			ServerHeartbeatSucceeded: func(e *event.ServerHeartbeatSucceededEvent) {
-				fmt.Println("Server heartbeat succeeded:", e.ConnectionID, e.Duration, e.Reply)
-			},
-		}
+		err := mt.Client.Ping(context.Background(), readpref.Primary())
+		assert.NoError(mt, err, "expected no error, got: %v", err)
 
-		commandMonitor := &event.CommandMonitor{
-			Started: func(_ context.Context, cse *event.CommandStartedEvent) {
-				fmt.Println("Command started:", cse.CommandName, cse.Command)
-			},
-			Succeeded: func(_ context.Context, cse *event.CommandSucceededEvent) {
-				fmt.Println("Command succeeded:", cse.CommandName, cse.Reply)
-			},
-		}
+		_, err = client.Database("test").Collection("test").InsertOne(context.Background(), bson.D{{"advance", "$clusterTime"}})
+		require.NoError(mt, err, "expected no error inserting document, got: %v", err)
 
-		opts := options.Client().
-			ApplyURI(mtest.ClusterURI()).
-			SetHosts([]string{mtest.ClusterConnString().Hosts[0]}).
-			SetDirect(true).
-			SetHeartbeatInterval(500 * time.Millisecond). // Minimum interval
-			SetServerMonitor(serverMonitor).
-			SetMonitor(commandMonitor)
+		atomic.StoreUint32(&clusterTimeAdvanced, 1)
+		<-heartbeatStarted
+		<-heartbeatSucceeded
 
-		client, err := mongo.Connect(opts)
-		require.NoError(mt, err, "expected no error connecting to client, got: %v", err)
-		defer func() {
-			err = client.Disconnect(context.Background())
-			require.NoError(mt, err, "expected no error disconnecting client, got: %v", err)
-		}()
+		err = mt.Client.Ping(context.Background(), readpref.Primary())
+		require.NoError(mt, err, "expected no error, got: %v", err)
 
-		err = client.Ping(context.Background(), readpref.Primary())
-		assert.NoError(mt, err, "expected no error, got: %v", err)
-		// mt.Fatal(mtest.ClusterURI(), mtest.ClusterConnString().Hosts, mtest.GlobalTopology().Description().Servers)
+		succeededEvents := mt.GetAllSucceededEvents()
+		require.Len(mt, succeededEvents, 2, "expected 2 succeeded events, got: %v", len(succeededEvents))
+		require.Equal(mt, "ping", succeededEvents[0].CommandName, "expected first command to be ping, got: %v", succeededEvents[0].CommandName)
+		initialClusterTime, err := succeededEvents[0].Reply.LookupErr("$clusterTime")
+		require.NoError(mt, err, "$clusterTime not found in response")
+
+		startedEvents := mt.GetAllStartedEvents()
+		require.Len(mt, startedEvents, 2, "expected 2 started events, got: %v", len(startedEvents))
+		require.Equal(mt, "ping", startedEvents[1].CommandName, "expected second command to be ping, got: %v", startedEvents[1].CommandName)
+		currentClusterTime, err := startedEvents[1].Command.LookupErr("$clusterTime")
+		require.NoError(mt, err, "$clusterTime not found in commane")
+		assert.Equal(mt, initialClusterTime, currentClusterTime, "expected same cluster time, got %v and %v", initialClusterTime, currentClusterTime)
 	})
 }