Skip to content

Commit b6631da

Browse files
jstewart148jstewart148
committed
Add getters for txnNumber and lsid fields to ChangeStreamDocument
JAVA-3225
1 parent f0ad2a2 commit b6631da

File tree

3 files changed

+137
-3
lines changed

3 files changed

+137
-3
lines changed

‎driver-core/src/main/com/mongodb/client/model/changestream/ChangeStreamDocument.java‎

Lines changed: 71 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
import com.mongodb.assertions.Assertions;
2121
import com.mongodb.lang.Nullable;
2222
import org.bson.BsonDocument;
23+
import org.bson.BsonInt64;
2324
import org.bson.BsonString;
2425
import org.bson.BsonTimestamp;
2526
import org.bson.codecs.Codec;
@@ -49,6 +50,8 @@ public final class ChangeStreamDocument<TDocument> {
4950
private final BsonTimestamp clusterTime;
5051
private final OperationType operationType;
5152
private final UpdateDescription updateDescription;
53+
private final BsonInt64 txnNumber;
54+
private final BsonDocument lsid;
5255

5356
/**
5457
* Creates a new instance
@@ -135,7 +138,7 @@ public ChangeStreamDocument(@BsonProperty("resumeToken") final BsonDocument resu
135138
*
136139
* @since 3.11
137140
*/
138-
@BsonCreator
141+
@Deprecated
139142
public ChangeStreamDocument(@BsonProperty("operationType") final OperationType operationType,
140143
@BsonProperty("resumeToken") final BsonDocument resumeToken,
141144
@Nullable @BsonProperty("ns") final BsonDocument namespaceDocument,
@@ -144,6 +147,37 @@ public ChangeStreamDocument(@BsonProperty("operationType") final OperationType o
144147
@Nullable @BsonProperty("documentKey") final BsonDocument documentKey,
145148
@Nullable @BsonProperty("clusterTime") final BsonTimestamp clusterTime,
146149
@Nullable @BsonProperty("updateDescription") final UpdateDescription updateDescription) {
150+
this(operationType, resumeToken, namespaceDocument, destinationNamespaceDocument, fullDocument, documentKey, clusterTime,
151+
updateDescription, null, null);
152+
}
153+
154+
/**
155+
* Creates a new instance
156+
*
157+
* @param operationType the operation type
158+
* @param resumeToken the resume token
159+
* @param namespaceDocument the BsonDocument representing the namespace
160+
* @param destinationNamespaceDocument the BsonDocument representing the destinatation namespace
161+
* @param fullDocument the full document
162+
* @param documentKey a document containing the _id of the changed document
163+
* @param clusterTime the cluster time at which the change occured
164+
* @param updateDescription the update description
165+
* @param txnNumber the transaction number
166+
* @param lsid the identifier for the session associated with the transaction
167+
*
168+
* @since 3.11
169+
*/
170+
@BsonCreator
171+
public ChangeStreamDocument(@BsonProperty("operationType") final OperationType operationType,
172+
@BsonProperty("resumeToken") final BsonDocument resumeToken,
173+
@Nullable @BsonProperty("ns") final BsonDocument namespaceDocument,
174+
@Nullable @BsonProperty("to") final BsonDocument destinationNamespaceDocument,
175+
@Nullable @BsonProperty("fullDocument") final TDocument fullDocument,
176+
@Nullable @BsonProperty("documentKey") final BsonDocument documentKey,
177+
@Nullable @BsonProperty("clusterTime") final BsonTimestamp clusterTime,
178+
@Nullable @BsonProperty("updateDescription") final UpdateDescription updateDescription,
179+
@Nullable @BsonProperty("txnNumber") final BsonInt64 txnNumber,
180+
@Nullable @BsonProperty("lsid") final BsonDocument lsid) {
147181
this.resumeToken = resumeToken;
148182
this.namespaceDocument = namespaceDocument;
149183
this.destinationNamespaceDocument = destinationNamespaceDocument;
@@ -152,6 +186,8 @@ public ChangeStreamDocument(@BsonProperty("operationType") final OperationType o
152186
this.clusterTime = clusterTime;
153187
this.operationType = operationType;
154188
this.updateDescription = updateDescription;
189+
this.txnNumber = txnNumber;
190+
this.lsid = lsid;
155191
}
156192

157193
private static BsonDocument namespaceToDocument(final MongoNamespace namespace) {
@@ -321,6 +357,30 @@ public UpdateDescription getUpdateDescription() {
321357
return updateDescription;
322358
}
323359

360+
/**
361+
* Returns the transaction number
362+
*
363+
* @return the transaction number, or null if not part of a multi-document transaction
364+
* @since 3.11
365+
* @mongodb.server.release 4.0
366+
*/
367+
@Nullable
368+
public BsonInt64 getTxnNumber() {
369+
return txnNumber;
370+
}
371+
372+
/**
373+
* Returns the identifier for the session associated with the transaction
374+
*
375+
* @return the lsid, or null if not part of a multi-document transaction
376+
* @since 3.11
377+
* @mongodb.server.release 4.0
378+
*/
379+
@Nullable
380+
public BsonDocument getLsid() {
381+
return lsid;
382+
}
383+
324384
/**
325385
* Creates the codec for the specific ChangeStreamOutput type
326386
*
@@ -371,6 +431,12 @@ public boolean equals(final Object o) {
371431
if (updateDescription != null ? !updateDescription.equals(that.updateDescription) : that.updateDescription != null) {
372432
return false;
373433
}
434+
if (txnNumber != null ? !txnNumber.equals(that.txnNumber) : that.txnNumber != null) {
435+
return false;
436+
}
437+
if (lsid != null ? !lsid.equals(that.lsid) : that.lsid != null) {
438+
return false;
439+
}
374440

375441
return true;
376442
}
@@ -385,6 +451,8 @@ public int hashCode() {
385451
result = 31 * result + (clusterTime != null ? clusterTime.hashCode() : 0);
386452
result = 31 * result + (operationType != null ? operationType.hashCode() : 0);
387453
result = 31 * result + (updateDescription != null ? updateDescription.hashCode() : 0);
454+
result = 31 * result + (txnNumber != null ? txnNumber.hashCode() : 0);
455+
result = 31 * result + (lsid != null ? lsid.hashCode() : 0);
388456
return result;
389457
}
390458

@@ -399,6 +467,8 @@ public String toString() {
399467
+ ", documentKey=" + documentKey
400468
+ ", clusterTime=" + clusterTime
401469
+ ", updateDescription=" + updateDescription
470+
+ ", txnNumber=" + txnNumber
471+
+ ", lsid=" + lsid
402472
+ "}";
403473
}
404474
}

‎driver-core/src/test/unit/com/mongodb/client/model/changestream/ChangeStreamDocumentCodecSpecification.groovy‎

Lines changed: 39 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ package com.mongodb.client.model.changestream
1919
import org.bson.BsonDocument
2020
import org.bson.BsonDocumentReader
2121
import org.bson.BsonDocumentWriter
22+
import org.bson.BsonInt64
2223
import org.bson.BsonReader
2324
import org.bson.BsonTimestamp
2425
import org.bson.Document
@@ -142,8 +143,19 @@ class ChangeStreamDocumentCodecSpecification extends Specification {
142143
,
143144
null
144145
),
146+
new ChangeStreamDocument<Document>(OperationType.INSERT,
147+
BsonDocument.parse('{token: true}'),
148+
BsonDocument.parse('{db: "engineering", coll: "users"}'),
149+
null,
150+
Document.parse('{_id: 1, userName: "alice123", name: "Alice"}'),
151+
BsonDocument.parse('{userName: "alice123", _id: 1}'),
152+
new BsonTimestamp(1234, 2),
153+
null,
154+
new BsonInt64(1),
155+
BsonDocument.parse('{id: 1, uid: 2}')
156+
),
145157
]
146-
clazz << [Document, Document, Document, Document, Document, Document, Document, Document, Document
158+
clazz << [Document, Document, Document, Document, Document, Document, Document, Document, Document, Document
147159
]
148160
json << [
149161
'''
@@ -286,7 +298,32 @@ class ChangeStreamDocumentCodecSpecification extends Specification {
286298
operationType: 'invalidate',
287299
clusterTime: { "$timestamp" : { "t" : 1234, "i" : 2 } }
288300
}
289-
'''
301+
''',
302+
'''
303+
{
304+
_id: { token : true },
305+
operationType: 'insert',
306+
clusterTime: { "$timestamp" : { "t" : 1234, "i" : 2 } },
307+
ns: {
308+
db: 'engineering',
309+
coll: 'users'
310+
},
311+
documentKey: {
312+
userName: 'alice123',
313+
_id: 1
314+
},
315+
fullDocument: {
316+
_id: 1,
317+
userName: 'alice123',
318+
name: 'Alice'
319+
},
320+
txnNumber: NumberLong('1'),
321+
lsid: {
322+
id: 1,
323+
uid: 2
324+
}
325+
}
326+
''',
290327
]
291328
}
292329
}

‎driver-core/src/test/unit/com/mongodb/client/model/changestream/ChangeStreamDocumentSpecification.groovy‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ package com.mongodb.client.model.changestream
1818

1919
import com.mongodb.MongoNamespace
2020
import org.bson.BsonDocument
21+
import org.bson.BsonInt64
2122
import org.bson.BsonTimestamp
2223
import org.bson.RawBsonDocument
2324
import spock.lang.Specification
@@ -36,6 +37,8 @@ class ChangeStreamDocumentSpecification extends Specification {
3637
def clusterTime = new BsonTimestamp(1234, 2)
3738
def operationType = OperationType.UPDATE
3839
def updateDesc = new UpdateDescription(['a', 'b'], BsonDocument.parse('{c: 1}'))
40+
def txnNumber = new BsonInt64(1);
41+
def lsid = BsonDocument.parse('{id: 1, uid: 1}');
3942

4043
when:
4144
def changeStreamDocument = new ChangeStreamDocument<BsonDocument>(operationType, resumeToken, namespaceDocument,
@@ -53,6 +56,28 @@ class ChangeStreamDocumentSpecification extends Specification {
5356
changeStreamDocument.getOperationType() == operationType
5457
changeStreamDocument.getUpdateDescription() == updateDesc
5558
changeStreamDocument.getDatabaseName() == namespace.getDatabaseName()
59+
changeStreamDocument.getTxnNumber() == null
60+
changeStreamDocument.getLsid() == null
61+
62+
when:
63+
def changeStreamDocumentWithTxnInfo = new ChangeStreamDocument<BsonDocument>(operationType, resumeToken,
64+
namespaceDocument, destinationNamespaceDocument, fullDocument, documentKey, clusterTime, updateDesc,
65+
txnNumber, lsid)
66+
67+
then:
68+
changeStreamDocumentWithTxnInfo.getResumeToken() == resumeToken
69+
changeStreamDocumentWithTxnInfo.getFullDocument() == fullDocument
70+
changeStreamDocumentWithTxnInfo.getDocumentKey() == documentKey
71+
changeStreamDocumentWithTxnInfo.getClusterTime() == clusterTime
72+
changeStreamDocumentWithTxnInfo.getNamespace() == namespace
73+
changeStreamDocumentWithTxnInfo.getNamespaceDocument() == namespaceDocument
74+
changeStreamDocumentWithTxnInfo.getDestinationNamespace() == destinationNamespace
75+
changeStreamDocumentWithTxnInfo.getDestinationNamespaceDocument() == destinationNamespaceDocument
76+
changeStreamDocumentWithTxnInfo.getOperationType() == operationType
77+
changeStreamDocumentWithTxnInfo.getUpdateDescription() == updateDesc
78+
changeStreamDocumentWithTxnInfo.getDatabaseName() == namespace.getDatabaseName()
79+
changeStreamDocumentWithTxnInfo.getTxnNumber() == txnNumber
80+
changeStreamDocumentWithTxnInfo.getLsid() == lsid
5681

5782
when:
5883
def changeStreamDocumentFromNamespace = new ChangeStreamDocument<BsonDocument>(resumeToken, namespace, fullDocument,
@@ -69,6 +94,8 @@ class ChangeStreamDocumentSpecification extends Specification {
6994
changeStreamDocumentFromNamespace.getDatabaseName() == namespace.getDatabaseName()
7095
changeStreamDocumentFromNamespace.getNamespaceDocument() == namespaceDocument
7196
changeStreamDocumentFromNamespace.getDestinationNamespaceDocument() == null
97+
changeStreamDocumentFromNamespace.getTxnNumber() == null
98+
changeStreamDocumentFromNamespace.getLsid() == null
7299
}
73100

74101
def 'should handle null namespace correctly'() {

0 commit comments

Comments
 (0)