读取隔离性、一致性和新近度

 /* Use a causally-consistent session to run some operations. */
 wc = mongoc_write_concern_new ();
 mongoc_write_concern_set_wmajority (wc, 1000);
 mongoc_collection_set_write_concern (coll, wc);
 rc = mongoc_read_concern_new ();
 mongoc_read_concern_set_level (rc, MONGOC_READ_CONCERN_LEVEL_MAJORITY);
 mongoc_collection_set_read_concern (coll, rc);
 session_opts = mongoc_session_opts_new ();
 mongoc_session_opts_set_causal_consistency (session_opts, true);
 session1 = mongoc_client_start_session (client, session_opts, &error);
 if (!session1) {
    fprintf (stderr, "couldn't start session: %s\n", error.message);
    goto cleanup;
 }
 /* Run an update_one with our causally-consistent session. */
 update_opts = bson_new ();
 res = mongoc_client_session_append (session1, update_opts, &error);
 if (!res) {
    fprintf (stderr, "couldn't add session to opts: %s\n", error.message);
    goto cleanup;
 }
 query = BCON_NEW ("sku", "111");
 update = BCON_NEW ("$set", "{", "end",
      BCON_DATE_TIME (bson_get_monotonic_time ()), "}");
 res = mongoc_collection_update_one (coll,
		       query,
		       update,
		       update_opts,
		       NULL, /* reply */
		       &error);
 if (!res) {
    fprintf (stderr, "update failed: %s\n", error.message);
    goto cleanup;
 }
 /* Run an insert with our causally-consistent session */
 insert_opts = bson_new ();
 res = mongoc_client_session_append (session1, insert_opts, &error);
 if (!res) {
    fprintf (stderr, "couldn't add session to opts: %s\n", error.message);
    goto cleanup;
 }
 insert = BCON_NEW ("sku", "nuts-111", "name", "Pecans",
      "start", BCON_DATE_TIME (bson_get_monotonic_time ()));
 res = mongoc_collection_insert_one (coll, insert, insert_opts, NULL, &error);
 if (!res) {
    fprintf (stderr, "insert failed: %s\n", error.message);
    goto cleanup;
 }

using (var session1 = client.StartSession(new ClientSessionOptions { CausalConsistency = true }))
{
    var currentDate = DateTime.UtcNow.Date;
    var items = client.GetDatabase(
        "test",
        new MongoDatabaseSettings
        {
            ReadConcern = ReadConcern.Majority,
            WriteConcern = new WriteConcern(
                    WriteConcern.WMode.Majority,
                    TimeSpan.FromMilliseconds(1000))
        })
        .GetCollection<BsonDocument>("items");
    items.UpdateOne(session1,
        Builders<BsonDocument>.Filter.And(
            Builders<BsonDocument>.Filter.Eq("sku", "111"),
            Builders<BsonDocument>.Filter.Eq("end", BsonNull.Value)),
        Builders<BsonDocument>.Update.Set("end", currentDate));
    items.InsertOne(session1, new BsonDocument
    {
        {"sku", "nuts-111"},
        {"name", "Pecans"},
        {"start", currentDate}
    });
}

// Example 1: Use a causally consistent session to ensure that the update occurs before the insert.
ClientSession session1 = client.startSession(ClientSessionOptions.builder().causallyConsistent(true).build());
Date currentDate = new Date();
MongoCollection<Document> items = client.getDatabase("test")
        .withReadConcern(ReadConcern.MAJORITY)
        .withWriteConcern(WriteConcern.MAJORITY.withWTimeout(1000, TimeUnit.MILLISECONDS))
        .getCollection("test");
items.updateOne(session1, eq("sku", "111"), set("end", currentDate));
Document document = new Document("sku", "nuts-111")
        .append("name", "Pecans")
        .append("start", currentDate);
items.insertOne(session1, document);

  async with await client.start_session(causal_consistency=True) as s1:
      current_date = datetime.datetime.today()
      items = client.get_database(
          "test",
          read_concern=ReadConcern("majority"),
          write_concern=WriteConcern("majority", wtimeout=1000),
      ).items
      await items.update_one(
          {"sku": "111", "end": None}, {"$set": {"end": current_date}}, session=s1
      )
      await items.insert_one(
          {"sku": "nuts-111", "name": "Pecans", "start": current_date}, session=s1
      )

my $s1 = $conn->start_session({ causalConsistency => 1 });
$items = $conn->get_database(
    "test", {
        read_concern => { level => 'majority' },
        write_concern => { w => 'majority', wtimeout => 10000 },
    }
)->get_collection("items");
$items->update_one(
    {
        sku => 111,
        end  => undef
    },
    {
        '$set' => { end => $current_date}
    },
    {
        session => $s1
    }
);
$items->insert_one(
    {
        sku => "nuts-111",
        name  => "Pecans",
        start => $current_date
    },
    {
        session => $s1
    }
);

$items = $client->selectDatabase(
    'test',
    [
        'readConcern' => new \MongoDB\Driver\ReadConcern(\MongoDB\Driver\ReadConcern::MAJORITY),
        'writeConcern' => new \MongoDB\Driver\WriteConcern(\MongoDB\Driver\WriteConcern::MAJORITY, 1000),
    ],
)->items;
$s1 = $client->startSession(
    ['causalConsistency' => true],
);
$currentDate = new \MongoDB\BSON\UTCDateTime();
$items->updateOne(
    ['sku' => '111', 'end' => ['$exists' => false]],
    ['$set' => ['end' => $currentDate]],
    ['session' => $s1],
);
$items->insertOne(
    ['sku' => '111-nuts', 'name' => 'Pecans', 'start' => $currentDate],
    ['session' => $s1],
);

with client.start_session(causal_consistency=True) as s1:
    current_date = datetime.datetime.today()
    items = client.get_database(
        "test",
        read_concern=ReadConcern("majority"),
        write_concern=WriteConcern("majority", wtimeout=1000),
    ).items
    items.update_one(
        {"sku": "111", "end": None}, {"$set": {"end": current_date}}, session=s1
    )
    items.insert_one(
        {"sku": "nuts-111", "name": "Pecans", "start": current_date}, session=s1
    )

let s1 = client1.startSession(options: ClientSessionOptions(causalConsistency: true))
let currentDate = Date()
var dbOptions = MongoDatabaseOptions(
    readConcern: .majority,
    writeConcern: try .majority(wtimeoutMS: 1000)
)
let items = client1.db("test", options: dbOptions).collection("items")
let result1 = items.updateOne(
    filter: ["sku": "111", "end": .null],
    update: ["$set": ["end": .datetime(currentDate)]],
    session: s1
).flatMap { _ in
    items.insertOne(["sku": "nuts-111", "name": "Pecans", "start": .datetime(currentDate)], session: s1)
}

let s1 = client1.startSession(options: ClientSessionOptions(causalConsistency: true))
let currentDate = Date()
var dbOptions = MongoDatabaseOptions(
    readConcern: .majority,
    writeConcern: try .majority(wtimeoutMS: 1000)
)
let items = client1.db("test", options: dbOptions).collection("items")
try items.updateOne(
    filter: ["sku": "111", "end": .null],
    update: ["$set": ["end": .datetime(currentDate)]],
    session: s1
)
try items.insertOne(["sku": "nuts-111", "name": "Pecans", "start": .datetime(currentDate)], session: s1)

 /* Make a new session, session2, and make it causally-consistent
  * with session1, so that session2 will read session1's writes. */
 session2 = mongoc_client_start_session (client, session_opts, &error);
 if (!session2) {
    fprintf (stderr, "couldn't start session: %s\n", error.message);
    goto cleanup;
 }
 /* Set the cluster time for session2 to session1's cluster time */
 cluster_time = mongoc_client_session_get_cluster_time (session1);
 mongoc_client_session_advance_cluster_time (session2, cluster_time);
 /* Set the operation time for session2 to session2's operation time */
 mongoc_client_session_get_operation_time (session1, &timestamp, &increment);
 mongoc_client_session_advance_operation_time (session2,
				 timestamp,
				 increment);
 /* Run a find on session2, which should now find all writes done
  * inside of session1 */
 find_opts = bson_new ();
 res = mongoc_client_session_append (session2, find_opts, &error);
 if (!res) {
    fprintf (stderr, "couldn't add session to opts: %s\n", error.message);
    goto cleanup;
 }
 find_query = BCON_NEW ("end", BCON_NULL);
 read_prefs = mongoc_read_prefs_new (MONGOC_READ_SECONDARY);
 cursor = mongoc_collection_find_with_opts (coll,
			      query,
			      find_opts,
			      read_prefs);
 while (mongoc_cursor_next (cursor, &result)) {
    json = bson_as_json (result, NULL);
    fprintf (stdout, "Document: %s\n", json);
    bson_free (json);
 }
 if (mongoc_cursor_error (cursor, &error)) {
    fprintf (stderr, "cursor failure: %s\n", error.message);
    goto cleanup;
 }

using (var session2 = client.StartSession(new ClientSessionOptions { CausalConsistency = true }))
{
    session2.AdvanceClusterTime(session1.ClusterTime);
    session2.AdvanceOperationTime(session1.OperationTime);
    var items = client.GetDatabase(
        "test",
        new MongoDatabaseSettings
        {
            ReadPreference = ReadPreference.Secondary,
            ReadConcern = ReadConcern.Majority,
            WriteConcern = new WriteConcern(WriteConcern.WMode.Majority, TimeSpan.FromMilliseconds(1000))
        })
        .GetCollection<BsonDocument>("items");
    var filter = Builders<BsonDocument>.Filter.Eq("end", BsonNull.Value);
    foreach (var item in items.Find(session2, filter).ToEnumerable())
    {
        // process item
    }
}

// Example 2: Advance the cluster time and the operation time to that of the other session to ensure that
// this client is causally consistent with the other session and read after the two writes.
ClientSession session2 = client.startSession(ClientSessionOptions.builder().causallyConsistent(true).build());
session2.advanceClusterTime(session1.getClusterTime());
session2.advanceOperationTime(session1.getOperationTime());
items = client.getDatabase("test")
        .withReadPreference(ReadPreference.secondary())
        .withReadConcern(ReadConcern.MAJORITY)
        .withWriteConcern(WriteConcern.MAJORITY.withWTimeout(1000, TimeUnit.MILLISECONDS))
        .getCollection("items");
for (Document item: items.find(session2, eq("end", BsonNull.VALUE))) {
    System.out.println(item);
}

  async with await client.start_session(causal_consistency=True) as s2:
      s2.advance_cluster_time(s1.cluster_time)
      s2.advance_operation_time(s1.operation_time)
      items = client.get_database(
          "test",
          read_preference=ReadPreference.SECONDARY,
          read_concern=ReadConcern("majority"),
          write_concern=WriteConcern("majority", wtimeout=1000),
      ).items
      async for item in items.find({"end": None}, session=s2):
          print(item)

my $s2 = $conn->start_session({ causalConsistency => 1 });
$s2->advance_cluster_time( $s1->cluster_time );
$s2->advance_operation_time( $s1->operation_time );
$items = $conn->get_database(
    "test", {
        read_preference => 'secondary',
        read_concern => { level => 'majority' },
        write_concern => { w => 'majority', wtimeout => 10000 },
    }
)->get_collection("items");
$cursor = $items->find( { end => undef }, { session => $s2 } );
for my $item ( $cursor->all ) {
    say join(" ", %$item);
}

$s2 = $client->startSession(
    ['causalConsistency' => true],
);
$s2->advanceClusterTime($s1->getClusterTime());
$s2->advanceOperationTime($s1->getOperationTime());
$items = $client->selectDatabase(
    'test',
    [
        'readPreference' => new \MongoDB\Driver\ReadPreference(\MongoDB\Driver\ReadPreference::SECONDARY),
        'readConcern' => new \MongoDB\Driver\ReadConcern(\MongoDB\Driver\ReadConcern::MAJORITY),
        'writeConcern' => new \MongoDB\Driver\WriteConcern(\MongoDB\Driver\WriteConcern::MAJORITY, 1000),
    ],
)->items;
$result = $items->find(
    ['end' => ['$exists' => false]],
    ['session' => $s2],
);
foreach ($result as $item) {
    var_dump($item);
}

with client.start_session(causal_consistency=True) as s2:
    s2.advance_cluster_time(s1.cluster_time)
    s2.advance_operation_time(s1.operation_time)
    items = client.get_database(
        "test",
        read_preference=ReadPreference.SECONDARY,
        read_concern=ReadConcern("majority"),
        write_concern=WriteConcern("majority", wtimeout=1000),
    ).items
    for item in items.find({"end": None}, session=s2):
        print(item)

let options = ClientSessionOptions(causalConsistency: true)
let result2: EventLoopFuture<Void> = client2.withSession(options: options) { s2 in
    // The cluster and operation times are guaranteed to be non-nil since we already used s1 for operations above.
    s2.advanceClusterTime(to: s1.clusterTime!)
    s2.advanceOperationTime(to: s1.operationTime!)
    dbOptions.readPreference = .secondary
    let items2 = client2.db("test", options: dbOptions).collection("items")
    return items2.find(["end": .null], session: s2).flatMap { cursor in
        cursor.forEach { item in
            print(item)
        }
    }
}

try client2.withSession(options: ClientSessionOptions(causalConsistency: true)) { s2 in
    // The cluster and operation times are guaranteed to be non-nil since we already used s1 for operations above.
    s2.advanceClusterTime(to: s1.clusterTime!)
    s2.advanceOperationTime(to: s1.operationTime!)
    dbOptions.readPreference = .secondary
    let items2 = client2.db("test", options: dbOptions).collection("items")
    for item in try items2.find(["end": .null], session: s2) {
        print(item)
    }
}