diff --git a/build.properties b/build.properties index a41535ce2f4..d9e30e5402c 100644 --- a/build.properties +++ b/build.properties @@ -18,10 +18,10 @@ javac.source=1.5 # IMPORTANT: MAKE SURE YOU CHANGE BOTH lib.version AND lib.version.osi.compat, ACCORDING TO THIS PATTERN # lib.version=2.8.0-SNAPSHOT ==> lib.version.osgi.compat=2.8.0.BUILD-SNAPSHOT -# lib.version=2.8.0-RC1 ==> lib.version.osgi.compat=2.8.0.RC1 -# lib.version=2.8.0 ==> lib.version.osgi.compat=2.8.0.RELEASE -lib.version=2.12.0-SNAPSHOT -lib.version.osgi.compat=2.12.0.BUILD-SNAPSHOT +# lib.version=2.8.0-rc1 ==> lib.version.osgi.compat=2.8.0.RC1 +# lib.version=2.8.0 ==> lib.version.osgi.compat=2.8.0.RELEASE +lib.version=2.12.0 +lib.version.osgi.compat=2.12.0.RELEASE compatibility.baseline.version=2.11.0 url.libbase=http://driver-downloads.mongodb.org/java diff --git a/config/codenarc/codenarc.xml b/config/codenarc/codenarc.xml index 4600961744e..3d2a7bf3920 100644 --- a/config/codenarc/codenarc.xml +++ b/config/codenarc/codenarc.xml @@ -56,7 +56,7 @@ - + @@ -65,10 +65,13 @@ + + + - + diff --git a/pom.xml b/pom.xml index 36bdca26246..2508733b228 100644 --- a/pom.xml +++ b/pom.xml @@ -8,7 +8,7 @@ driver developers who would rather use Maven than Ant as their build tool. mongo-java-driver bundle MongoDB Java Driver - 2.12.0-SNAPSHOT + 2.12.0 The MongoDB Java driver http://www.mongodb.org diff --git a/src/main/META-INF/MANIFEST.MF b/src/main/META-INF/MANIFEST.MF index ae440f8defe..ca2b8f3918d 100644 --- a/src/main/META-INF/MANIFEST.MF +++ b/src/main/META-INF/MANIFEST.MF @@ -4,7 +4,13 @@ Bundle-ManifestVersion: 2 Bundle-Name: MongoDB Java Driver Bundle-SymbolicName: org.mongodb.mongo-java-driver Bundle-Version: @VERSION@ -Import-Package: javax.management, javax.net, javax.net.ssl +Import-Package: javax.management, + javax.net, + javax.net.ssl, + javax.security.sasl, + javax.security.auth.login, + javax.security.auth, + org.ietf.jgss Export-Package: com.mongodb;version="@VERSION@", com.mongodb.io;version="@VERSION@", com.mongodb.util;version="@VERSION@", diff --git a/src/main/com/mongodb/AcknowledgedBulkWriteResult.java b/src/main/com/mongodb/AcknowledgedBulkWriteResult.java index 92ad51c26d8..1d7d2817819 100644 --- a/src/main/com/mongodb/AcknowledgedBulkWriteResult.java +++ b/src/main/com/mongodb/AcknowledgedBulkWriteResult.java @@ -25,11 +25,11 @@ class AcknowledgedBulkWriteResult extends BulkWriteResult { private int insertedCount; private int matchedCount; private int removedCount; - private int modifiedCount; + private Integer modifiedCount; private final List upserts; AcknowledgedBulkWriteResult(final int insertedCount, final int matchedCount, final int removedCount, - final int modifiedCount, final List upserts) { + final Integer modifiedCount, final List upserts) { this.insertedCount = insertedCount; this.matchedCount = matchedCount; this.removedCount = removedCount; @@ -41,7 +41,7 @@ class AcknowledgedBulkWriteResult extends BulkWriteResult { this(type, count, 0, upserts); } - AcknowledgedBulkWriteResult(final WriteRequest.Type type, final int count, final int modifiedCount, + AcknowledgedBulkWriteResult(final WriteRequest.Type type, final int count, final Integer modifiedCount, final List upserts) { this(type == WriteRequest.Type.INSERT ? count : 0, (type == WriteRequest.Type.UPDATE || type == WriteRequest.Type.REPLACE) ? count : 0, @@ -69,9 +69,20 @@ public int getRemovedCount() { return removedCount; } + @Override + public boolean isModifiedCountAvailable() { + return modifiedCount != null; + } + @Override public int getModifiedCount() { + if (modifiedCount == null) { + throw new UnsupportedOperationException("The modifiedCount is not available because at least one of the servers that was " + + "updated was not able to provide this information (the server is must be at least " + + "version 2.6"); + } return modifiedCount; + } @Override @@ -93,13 +104,13 @@ public boolean equals(final Object o) { if (insertedCount != that.insertedCount) { return false; } - if (modifiedCount != that.modifiedCount) { + if (matchedCount != that.matchedCount) { return false; } if (removedCount != that.removedCount) { return false; } - if (matchedCount != that.matchedCount) { + if (modifiedCount != null ? !modifiedCount.equals(that.modifiedCount) : that.modifiedCount != null) { return false; } if (!upserts.equals(that.upserts)) { @@ -111,11 +122,11 @@ public boolean equals(final Object o) { @Override public int hashCode() { - int result = upserts.hashCode(); - result = 31 * result + insertedCount; + int result = insertedCount; result = 31 * result + matchedCount; result = 31 * result + removedCount; - result = 31 * result + modifiedCount; + result = 31 * result + (modifiedCount != null ? modifiedCount.hashCode() : 0); + result = 31 * result + upserts.hashCode(); return result; } diff --git a/src/main/com/mongodb/AggregationOutput.java b/src/main/com/mongodb/AggregationOutput.java index 970eae5e6d8..725add30cc4 100644 --- a/src/main/com/mongodb/AggregationOutput.java +++ b/src/main/com/mongodb/AggregationOutput.java @@ -16,11 +16,12 @@ package com.mongodb; +@SuppressWarnings("deprecation") public class AggregationOutput { /** * returns an iterator to the results of the aggregation - * @return + * @return the results of the aggregation */ public Iterable results() { return _resultSet; @@ -28,42 +29,53 @@ public Iterable results() { /** * returns the command result of the aggregation - * @return + * @return the command result + * + * @deprecated there is no replacement for this method */ + @Deprecated public CommandResult getCommandResult(){ return _commandResult; } /** * returns the original aggregation command - * @return + * @return the command + * + * @deprecated there is no replacement for this method */ + @Deprecated public DBObject getCommand() { return _cmd; } /** * returns the address of the server used to execute the aggregation - * @return + * @return the server which executed the aggregation + * + * @deprecated there is no replacement for this method */ + @Deprecated public ServerAddress getServerUsed() { return _commandResult.getServerUsed(); } /** - * string representation of the aggregation command + * Constructs a new instance + * + * @param command the aggregation command + * @param commandResult the aggregation command result + * + * @deprecated there is no replacement for this constructor */ - public String toString(){ - return _commandResult.toString(); - } - @SuppressWarnings("unchecked") - public AggregationOutput(DBObject cmd, CommandResult raw) { - _commandResult = raw; - _cmd = cmd; + @Deprecated + public AggregationOutput(DBObject command, CommandResult commandResult) { + _commandResult = commandResult; + _cmd = command; - if(raw.containsField("result")) - _resultSet = (Iterable) raw.get( "result" ); + if(commandResult.containsField("result")) + _resultSet = (Iterable) commandResult.get( "result" ); else throw new IllegalArgumentException("result undefined"); } diff --git a/src/main/com/mongodb/AnyServerSelector.java b/src/main/com/mongodb/AnyServerSelector.java new file mode 100644 index 00000000000..c5cf66847be --- /dev/null +++ b/src/main/com/mongodb/AnyServerSelector.java @@ -0,0 +1,15 @@ +package com.mongodb; + +import java.util.List; + +class AnyServerSelector implements ServerSelector { + @Override + public List choose(final ClusterDescription clusterDescription) { + return clusterDescription.getAny(); + } + + @Override + public String toString() { + return "AnyServerSelector{}"; + } +} diff --git a/src/main/com/mongodb/BaseCluster.java b/src/main/com/mongodb/BaseCluster.java index 60d55fcd33d..b69792773f8 100644 --- a/src/main/com/mongodb/BaseCluster.java +++ b/src/main/com/mongodb/BaseCluster.java @@ -25,6 +25,7 @@ import java.util.logging.Logger; import static java.lang.String.format; +import static java.util.concurrent.TimeUnit.MILLISECONDS; import static java.util.concurrent.TimeUnit.NANOSECONDS; import static org.bson.util.Assertions.isTrue; import static org.bson.util.Assertions.notNull; @@ -76,27 +77,24 @@ public Server getServer(final ServerSelector serverSelector, final long maxWaitT } if (!curDescription.isConnecting()) { - throw new MongoServerSelectionException( - format("Unable to connect to any server that satisfies the selector " + - "%s", serverSelector)); + throw new MongoServerSelectionException(format("Unable to connect to any server that matches %s", serverSelector)); } final long timeout = endTime - System.nanoTime(); LOGGER.info(format("No server chosen by %s from cluster description %s. Waiting for %d ms before timing out", - serverSelector, curDescription, TimeUnit.MILLISECONDS.convert(timeout, NANOSECONDS))); + serverSelector, curDescription, MILLISECONDS.convert(timeout, NANOSECONDS))); if (!currentPhase.await(timeout, NANOSECONDS)) { - throw new MongoTimeoutException(format("Timed out while waiting for a server that satisfies the selector: %s " - + "after %d %s", serverSelector, timeout, NANOSECONDS)); + throw new MongoTimeoutException(format("Timed out while waiting for a server that matches %s after %d ms", + serverSelector, MILLISECONDS.convert(timeout, NANOSECONDS))); } currentPhase = phase.get(); curDescription = description; serverDescriptions = serverSelector.choose(curDescription); } } catch (InterruptedException e) { - throw new MongoInterruptedException(format("Interrupted while waiting for a server that satisfies server selector %s ", - serverSelector), e); + throw new MongoInterruptedException(format("Interrupted while waiting for a server that matches %s ", serverSelector), e); } } @@ -117,18 +115,18 @@ public ClusterDescription getDescription(final long maxWaitTime, final TimeUnit final long timeout = endTime - System.nanoTime(); LOGGER.info(format("Cluster description not yet available. Waiting for %d ms before timing out", - TimeUnit.MILLISECONDS.convert(timeout, NANOSECONDS))); + MILLISECONDS.convert(timeout, NANOSECONDS))); if (!currentPhase.await(timeout, NANOSECONDS)) { - throw new MongoTimeoutException(format("Timed out while waiting for the cluster description after waiting %d %s", - timeout, NANOSECONDS)); + throw new MongoTimeoutException(format("Timed out while waiting to connect after %d ms", + MILLISECONDS.convert(timeout, NANOSECONDS))); } currentPhase = phase.get(); curDescription = description; } return curDescription; } catch (InterruptedException e) { - throw new MongoInterruptedException(format("Interrupted while waiting for the cluster description"), e); + throw new MongoInterruptedException(format("Interrupted while waiting to connect"), e); } } diff --git a/src/main/com/mongodb/BaseWriteCommandMessage.java b/src/main/com/mongodb/BaseWriteCommandMessage.java index 25e9d159ac1..b5ca99f638c 100644 --- a/src/main/com/mongodb/BaseWriteCommandMessage.java +++ b/src/main/com/mongodb/BaseWriteCommandMessage.java @@ -87,9 +87,19 @@ private void writeCommandHeader(final OutputBuffer buffer) { protected abstract BaseWriteCommandMessage writeTheWrites(final OutputBuffer buffer, final int commandStartPosition, final BSONBinaryWriter writer); - protected boolean maximumCommandDocumentSizeExceeded(final OutputBuffer buffer, final int commandStartPosition) { - // Subtract 2 to account for the trailing 0x0 at the end of the enclosing array and command document - return buffer.getPosition() - commandStartPosition > getSettings().getMaxDocumentSize() + HEADROOM - 2; + protected boolean exceedsLimits(final int batchLength, final int batchItemCount) { + return (exceedsBatchLengthLimit(batchLength, batchItemCount) || exceedsBatchItemCountLimit(batchItemCount)); + } + + // make a special exception for a command with only a single item added to it. It's allowed to exceed maximum document size so that + // it's possible to, say, send a replacement document that is itself 16MB, which would push the size of the containing command + // document to be greater than the maximum document size. + private boolean exceedsBatchLengthLimit(final int batchLength, final int batchItemCount) { + return batchLength > getSettings().getMaxDocumentSize() && batchItemCount > 1; + } + + private boolean exceedsBatchItemCountLimit(final int batchItemCount) { + return batchItemCount > getSettings().getMaxWriteBatchSize(); } public abstract int getItemCount(); diff --git a/src/main/com/mongodb/BulkWriteBatchCombiner.java b/src/main/com/mongodb/BulkWriteBatchCombiner.java index c833ae8d671..aae4c50e487 100644 --- a/src/main/com/mongodb/BulkWriteBatchCombiner.java +++ b/src/main/com/mongodb/BulkWriteBatchCombiner.java @@ -33,7 +33,7 @@ class BulkWriteBatchCombiner { private int insertedCount; private int matchedCount; private int removedCount; - private int modifiedCount; + private Integer modifiedCount = 0; private final Set writeUpserts = new TreeSet(new Comparator() { @Override public int compare(final BulkWriteUpsert o1, final BulkWriteUpsert o2) { @@ -58,7 +58,11 @@ public void addResult(final BulkWriteResult result, final IndexMap indexMap) { insertedCount += result.getInsertedCount(); matchedCount += result.getMatchedCount(); removedCount += result.getRemovedCount(); - modifiedCount += result.getModifiedCount(); + if (result.isModifiedCountAvailable() && modifiedCount != null) { + modifiedCount += result.getModifiedCount(); + } else { + modifiedCount = null; + } mergeUpserts(result.getUpserts(), indexMap); } diff --git a/src/main/com/mongodb/BulkWriteError.java b/src/main/com/mongodb/BulkWriteError.java index 95e525039ef..fe1690b7e71 100644 --- a/src/main/com/mongodb/BulkWriteError.java +++ b/src/main/com/mongodb/BulkWriteError.java @@ -115,4 +115,14 @@ public int hashCode() { result = 31 * result + details.hashCode(); return result; } + + @Override + public String toString() { + return "BulkWriteError{" + + "index=" + index + + ", code=" + code + + ", message='" + message + '\'' + + ", details=" + details + + '}'; + } } diff --git a/src/main/com/mongodb/BulkWriteResult.java b/src/main/com/mongodb/BulkWriteResult.java index 465c0731f1f..a47958f6b61 100644 --- a/src/main/com/mongodb/BulkWriteResult.java +++ b/src/main/com/mongodb/BulkWriteResult.java @@ -65,15 +65,34 @@ public abstract class BulkWriteResult { */ public abstract int getRemovedCount(); + /** + * Returns true if the server was able to provide a count of modified documents. If this method returns false (which can happen if + * the server is not at least version 2.6) then the {@code getModifiedCount} method will throw {@code UnsupportedOperationException}. + * + * @return true if modifiedCount is available + * + * @throws UnacknowledgedWriteException if the write was unacknowledged. + * @see WriteConcern#UNACKNOWLEDGED + * @see #getModifiedCount() + */ + public abstract boolean isModifiedCountAvailable(); + /** * Returns the number of documents modified by updates or replacements in the write operation. This will only count documents that * were actually changed; for example, if you set the value of some field, and the field already has that value, * that will not count as a modification. + *

+ * If the server is not able to provide a count of modified documents (which can happen if the server is not at least version + * 2.6), then this method will throw an {@code UnsupportedOperationException} + *

* * @return the number of documents modified by the write operation * * @throws UnacknowledgedWriteException if the write was unacknowledged. + * @throws java.lang.UnsupportedOperationException if no modified count is available * @see WriteConcern#UNACKNOWLEDGED + * @see #isModifiedCountAvailable() + * @mongodb.server.release 2.6 */ public abstract int getModifiedCount(); diff --git a/src/main/com/mongodb/CompositeServerSelector.java b/src/main/com/mongodb/CompositeServerSelector.java index a95f2437dff..d8a8e401101 100644 --- a/src/main/com/mongodb/CompositeServerSelector.java +++ b/src/main/com/mongodb/CompositeServerSelector.java @@ -43,8 +43,8 @@ public List choose(final ClusterDescription clusterDescriptio @Override public String toString() { - return "CompositeServerSelector{" - + "serverSelectors=" + serverSelectors - + '}'; + return "{" + + "serverSelectors=" + serverSelectors + + '}'; } -} +} \ No newline at end of file diff --git a/src/main/com/mongodb/DBApiLayer.java b/src/main/com/mongodb/DBApiLayer.java index a4da4f5f5f3..2d78be81e81 100644 --- a/src/main/com/mongodb/DBApiLayer.java +++ b/src/main/com/mongodb/DBApiLayer.java @@ -25,8 +25,6 @@ import java.util.concurrent.ConcurrentLinkedQueue; import java.util.logging.Level; -import static java.util.Arrays.asList; - /** * Concrete extension of abstract {@code DB} class. * @@ -198,7 +196,7 @@ CommandResult doAuthenticate(MongoCredential credentials) { } private boolean useUserCommands(final DBPort port) { - return _connector.getServerDescription(port.getAddress()).getVersion().compareTo(new ServerVersion(asList(2, 5, 4))) >= 0; + return _connector.getServerDescription(port.getAddress()).getVersion().compareTo(new ServerVersion(2, 6)) >= 0; } void addDeadCursor(final DeadCursor deadCursor) { diff --git a/src/main/com/mongodb/DBCollection.java b/src/main/com/mongodb/DBCollection.java index d4b99802622..a23f75e1d86 100644 --- a/src/main/com/mongodb/DBCollection.java +++ b/src/main/com/mongodb/DBCollection.java @@ -574,6 +574,7 @@ public void createIndex( DBObject keys , DBObject options ){ * @mongodb.driver.manual /administration/indexes-creation/ Index Creation Tutorials * @deprecated use {@link #createIndex(DBObject, com.mongodb.DBObject)} the encoder is not used. */ + @Deprecated public abstract void createIndex(DBObject keys, DBObject options, DBEncoder encoder); /** diff --git a/src/main/com/mongodb/DBCollectionImpl.java b/src/main/com/mongodb/DBCollectionImpl.java index ee2f2424d50..492d039265e 100644 --- a/src/main/com/mongodb/DBCollectionImpl.java +++ b/src/main/com/mongodb/DBCollectionImpl.java @@ -44,6 +44,7 @@ import static java.util.Arrays.asList; import static org.bson.util.Assertions.isTrue; +@SuppressWarnings("deprecation") class DBCollectionImpl extends DBCollection { private final DBApiLayer db; private final String namespace; @@ -166,36 +167,24 @@ protected WriteResult insert(List list, boolean shouldApply , WriteCon if (encoder == null) encoder = DefaultDBEncoder.FACTORY.create(); - if ( willTrace() ) { for (DBObject o : list) { trace("save: " + namespace + " " + JSON.serialize(o)); } } - if ( shouldApply ){ - for (DBObject o : list) { - apply(o); - _checkObject(o, false, false); - Object id = o.get("_id"); - if (id instanceof ObjectId) { - ((ObjectId) id).notNew(); - } - } - } - DBPort port = db.getConnector().getPrimaryPort(); try { if (useWriteCommands(concern, port)) { try { - return translateBulkWriteResult(insertWithCommandProtocol(list, concern, encoder, port), INSERT, concern, + return translateBulkWriteResult(insertWithCommandProtocol(list, concern, encoder, port, shouldApply), INSERT, concern, port.getAddress()); } catch (BulkWriteException e) { throw translateBulkWriteException(e, INSERT); } } else { - return insertWithWriteProtocol(list, concern, encoder, port); + return insertWithWriteProtocol(list, concern, encoder, port, shouldApply); } } finally { db.getConnector().releasePort(port); @@ -342,11 +331,11 @@ public void createIndex(final DBObject keys, final DBObject options, DBEncoder e DBPort port = db.getConnector().getPrimaryPort(); try { - DBObject index = new BasicDBObject(); + DBObject index = defaultOptions(keys); index.putAll(options); index.put("key", keys); - if (connector.getServerDescription(port.getAddress()).getVersion().compareTo(new ServerVersion(asList(2, 5, 5))) >= 0) { + if (connector.getServerDescription(port.getAddress()).getVersion().compareTo(new ServerVersion(2, 6)) >= 0) { BasicDBObject createIndexes = new BasicDBObject("createIndexes", getName()); BasicDBList list = new BasicDBList(); @@ -365,7 +354,7 @@ public void createIndex(final DBObject keys, final DBObject options, DBEncoder e } } else { db.doGetCollection("system.indexes").insertWithWriteProtocol(asList(index), WriteConcern.SAFE, - DefaultDBEncoder.FACTORY.create(), port); + DefaultDBEncoder.FACTORY.create(), port, false); } } catch (IOException e) { throw new MongoException.Network("Operation on server " + port.getAddress() + " failed", e); @@ -376,19 +365,34 @@ public void createIndex(final DBObject keys, final DBObject options, DBEncoder e private BulkWriteResult insertWithCommandProtocol(final List list, final WriteConcern writeConcern, final DBEncoder encoder, - final DBPort port) { + final DBPort port, final boolean shouldApply) { + if ( shouldApply ){ + applyRulesForInsert(list); + } + BaseWriteCommandMessage message = new InsertCommandMessage(getNamespace(), writeConcern, list, DefaultDBEncoder.FACTORY.create(), encoder, - getMessageSettings(port.getAddress())); + getMessageSettings(port)); return writeWithCommandProtocol(port, INSERT, message, writeConcern); } + private void applyRulesForInsert(final List list) { + for (DBObject o : list) { + _checkObject(o, false, false); + apply(o); + Object id = o.get("_id"); + if (id instanceof ObjectId) { + ((ObjectId) id).notNew(); + } + } + } + private BulkWriteResult removeWithCommandProtocol(final List removeList, final WriteConcern writeConcern, final DBEncoder encoder, final DBPort port) { BaseWriteCommandMessage message = new DeleteCommandMessage(getNamespace(), writeConcern, removeList, DefaultDBEncoder.FACTORY.create(), encoder, - getMessageSettings(port.getAddress())); + getMessageSettings(port)); return writeWithCommandProtocol(port, REMOVE, message, writeConcern); } @@ -398,7 +402,7 @@ private BulkWriteResult updateWithCommandProtocol(final List upda final DBEncoder encoder, final DBPort port) { BaseWriteCommandMessage message = new UpdateCommandMessage(getNamespace(), writeConcern, updates, DefaultDBEncoder.FACTORY.create(), encoder, - getMessageSettings(port.getAddress())); + getMessageSettings(port)); return writeWithCommandProtocol(port, UPDATE, message, writeConcern); } @@ -431,16 +435,22 @@ private BulkWriteResult writeWithCommandProtocol(final DBPort port, final WriteR private boolean useWriteCommands(final WriteConcern concern, final DBPort port) { return concern.callGetLastError() && - db.getConnector().getServerDescription(port.getAddress()).getVersion().compareTo(new ServerVersion(asList(2, 5, 5))) >= 0; + db.getConnector().getServerDescription(port.getAddress()).getVersion().compareTo(new ServerVersion(2, 6)) >= 0; } - private MessageSettings getMessageSettings(final ServerAddress address) { - ServerDescription serverDescription = db.getConnector().getServerDescription(address); - return MessageSettings.builder().maxDocumentSize(serverDescription.getMaxDocumentSize()).maxMessageSize(serverDescription - .getMaxMessageSize()) + private MessageSettings getMessageSettings(final DBPort port) { + ServerDescription serverDescription = db.getConnector().getServerDescription(port.getAddress()); + return MessageSettings.builder() + .maxDocumentSize(serverDescription.getMaxDocumentSize()) + .maxMessageSize(serverDescription.getMaxMessageSize()) + .maxWriteBatchSize(serverDescription.getMaxWriteBatchSize()) .build(); } + private int getMaxWriteBatchSize(final DBPort port) { + return db.getConnector().getServerDescription(port.getAddress()).getMaxWriteBatchSize(); + } + private MongoNamespace getNamespace() { return new MongoNamespace(getDB().getName(), getName()); } @@ -482,7 +492,11 @@ public CommandResult execute() throws IOException { private WriteResult insertWithWriteProtocol(final List list, final WriteConcern concern, final DBEncoder encoder, - final DBPort port) { + final DBPort port, final boolean shouldApply) { + if ( shouldApply ){ + applyRulesForInsert(list); + } + WriteResult last = null; int cur = 0; @@ -534,16 +548,16 @@ private Logger getLogger() { private class OrderedRunGenerator implements Iterable { private final List writeRequests; - private final DBPort port; private final WriteConcern writeConcern; private final DBEncoder encoder; + private final int maxBatchWriteSize; public OrderedRunGenerator(final List writeRequests, final WriteConcern writeConcern, final DBEncoder encoder, final DBPort port) { this.writeRequests = writeRequests; - this.port = port; this.writeConcern = writeConcern.continueOnError(false); this.encoder = encoder; + this.maxBatchWriteSize = getMaxWriteBatchSize(port); } @Override @@ -570,8 +584,7 @@ public Run next() { private int getStartIndexOfNextRun() { WriteRequest.Type type = writeRequests.get(curIndex).getType(); for (int i = curIndex; i < writeRequests.size(); i++) { - if (i == db.getConnector().getServerDescription(port.getAddress()).getMaxWriteBatchSize() - || writeRequests.get(i).getType() != type) { + if (i == curIndex + maxBatchWriteSize || writeRequests.get(i).getType() != type) { return i; } } @@ -589,16 +602,16 @@ public void remove() { private class UnorderedRunGenerator implements Iterable { private final List writeRequests; - private final DBPort port; private final WriteConcern writeConcern; private final DBEncoder encoder; + private final int maxBatchWriteSize; public UnorderedRunGenerator(final List writeRequests, final WriteConcern writeConcern, final DBEncoder encoder, final DBPort port) { this.writeRequests = writeRequests; - this.port = port; this.writeConcern = writeConcern.continueOnError(true); this.encoder = encoder; + this.maxBatchWriteSize = getMaxWriteBatchSize(port); } @Override @@ -629,7 +642,7 @@ public Run next() { } run.add(writeRequest, curIndex); curIndex++; - if (run.size() > db.getConnector().getServerDescription(port.getAddress()).getMaxWriteBatchSize()) { + if (run.size() == maxBatchWriteSize) { return runs.remove(run.type); } } @@ -704,6 +717,14 @@ private List getWriteRequestsAsModifyRequests() { } BulkWriteResult executeUpdates(final List updateRequests, final DBPort port) { + for (ModifyRequest request : updateRequests) { + for (String key : request.getUpdateDocument().keySet()) { + if (!key.startsWith("$")) { + throw new IllegalArgumentException("Update document keys must start with $: " + key); + } + } + } + return new RunExecutor(port) { @Override BulkWriteResult executeWriteCommandProtocol() { @@ -725,6 +746,10 @@ WriteRequest.Type getType() { } BulkWriteResult executeReplaces(final List replaceRequests, final DBPort port) { + for (ModifyRequest request : replaceRequests) { + _checkObject(request.getUpdateDocument(), false, false); + } + return new RunExecutor(port) { @Override BulkWriteResult executeWriteCommandProtocol() { @@ -773,7 +798,7 @@ BulkWriteResult executeWriteCommandProtocol() { for (InsertRequest cur : insertRequests) { documents.add(cur.getDocument()); } - return insertWithCommandProtocol(documents, writeConcern, encoder, port); + return insertWithCommandProtocol(documents, writeConcern, encoder, port, true); } @Override @@ -806,7 +831,7 @@ private abstract class RunExecutor { BulkWriteResult getResult(final WriteResult writeResult) { int count = getCount(writeResult); List upsertedItems = getUpsertedItems(writeResult); - int modifiedCount = (getType() == UPDATE || getType() == REPLACE) ? count - upsertedItems.size() : 0; + Integer modifiedCount = (getType() == UPDATE || getType() == REPLACE) ? null : 0; return new AcknowledgedBulkWriteResult(getType(), count - upsertedItems.size(), modifiedCount, upsertedItems); } @@ -826,10 +851,16 @@ private BulkWriteResult executeWriteProtocol() { WriteResult writeResult = executeWriteProtocol(i); if (writeConcern.callGetLastError()) { bulkWriteBatchCombiner.addResult(getResult(writeResult), indexMap); + // When a journaled write is requested but journaling is disabled, it's not thrown as an exception. + if (isWriteConcernError(writeResult.getLastError())) { + bulkWriteBatchCombiner.addWriteConcernErrorResult(getWriteConcernError(writeResult.getLastError())); + } } } catch (WriteConcernException writeException) { - if (isWriteConcernError(writeException)) { - bulkWriteBatchCombiner.addWriteConcernErrorResult(getWriteConcernError(writeException)); + if (isWriteConcernError(writeException.getCommandResult())) { + bulkWriteBatchCombiner.addResult(getResult(new WriteResult(writeException.getCommandResult(), writeConcern)), + indexMap); + bulkWriteBatchCombiner.addWriteConcernErrorResult(getWriteConcernError(writeException.getCommandResult())); } else { bulkWriteBatchCombiner.addWriteErrorResult(getBulkWriteError(writeException), indexMap); } @@ -846,10 +877,6 @@ private int getCount(final WriteResult writeResult) { return getType() == INSERT ? 1 : writeResult.getN(); } - private boolean isWriteConcernError(final WriteConcernException writeException) { - return writeException.getCommandResult().get("wtimeout") != null; - } - List getUpsertedItems(final WriteResult writeResult) { return writeResult.getUpsertedId() == null ? Collections.emptyList() @@ -863,9 +890,26 @@ private BulkWriteError getBulkWriteError(final WriteConcernException writeExcept 0); } - private WriteConcernError getWriteConcernError(final WriteConcernException writeException) { - return new WriteConcernError(writeException.getCode(), writeException.getCommandResult().getString("err"), - getErrorResponseDetails(writeException.getCommandResult())); + // Accommodating GLE representation of write concern errors + private boolean isWriteConcernError(final CommandResult commandResult) { + return commandResult.get("wtimeout") != null || commandResult.get("wnote") != null || commandResult.get("jnote") != null; + } + + private WriteConcernError getWriteConcernError(final CommandResult commandResult) { + return new WriteConcernError(commandResult.getCode(), getWriteConcernErrorMessage(commandResult), + getErrorResponseDetails(commandResult)); + } + + // GLE uses jnote and wnote as alternative ways or reporting write concern errors + private String getWriteConcernErrorMessage(final CommandResult commandResult) { + String errorMessage = commandResult.getString("jnote"); + if (errorMessage == null) { + errorMessage = commandResult.getString("wnote"); + } + if (errorMessage == null) { + errorMessage = commandResult.getString("err"); + } + return errorMessage; } private DBObject getErrorResponseDetails(final DBObject response) { diff --git a/src/main/com/mongodb/DBPointer.java b/src/main/com/mongodb/DBPointer.java index 77b3be72a2e..d815f39c86c 100644 --- a/src/main/com/mongodb/DBPointer.java +++ b/src/main/com/mongodb/DBPointer.java @@ -26,7 +26,7 @@ @Deprecated public class DBPointer extends DBRefBase { - static final boolean D = Boolean.getBoolean( "DEBUG.DBPOINTER" ); + private static final long serialVersionUID = -1977838613745447826L; /** * CTOR used for testing BSON encoding. Otherwise @@ -41,7 +41,7 @@ public DBPointer(String ns, ObjectId id) { } DBPointer( DBObject parent , String fieldName , DB db , String ns , ObjectId id ){ - super(db, ns, (Object)id); + super(db, ns, id); _parent = parent; _fieldName = fieldName; diff --git a/src/main/com/mongodb/DBTCPConnector.java b/src/main/com/mongodb/DBTCPConnector.java index 62ad5049a67..59b079974d4 100644 --- a/src/main/com/mongodb/DBTCPConnector.java +++ b/src/main/com/mongodb/DBTCPConnector.java @@ -20,7 +20,6 @@ import java.io.InterruptedIOException; import java.net.SocketTimeoutException; import java.util.ArrayList; -import java.util.Arrays; import java.util.List; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; @@ -31,7 +30,8 @@ import static com.mongodb.ClusterConnectionMode.Single; import static com.mongodb.ClusterType.ReplicaSet; import static com.mongodb.ClusterType.Sharded; -import static com.mongodb.MongoAuthority.Type.Direct; +import static com.mongodb.ClusterType.Unknown; +import static com.mongodb.MongoAuthority.Type.Set; import static java.util.concurrent.TimeUnit.MILLISECONDS; import static org.bson.util.Assertions.isTrue; @@ -52,7 +52,10 @@ public class DBTCPConnector implements DBConnector { private final MyPort _myPort = new MyPort(); - private ServerSelector prefixedServerSelector; + private final ClusterConnectionMode connectionMode; + + private ClusterType type = ClusterType.Unknown; + private MongosHAServerSelector mongosHAServerSelector; /** * @param mongo the Mongo instance @@ -60,6 +63,8 @@ public class DBTCPConnector implements DBConnector { */ public DBTCPConnector( Mongo mongo ) { _mongo = mongo; + connectionMode = _mongo.getAuthority().getType() == Set || _mongo.getMongoOptions().getRequiredReplicaSetName() != null ? + Multiple : Single; } public void start() { @@ -76,7 +81,8 @@ public void start() { Clusters.create(clusterId, ClusterSettings.builder() .hosts(_mongo.getAuthority().getServerAddresses()) - .mode(_mongo.getAuthority().getType() == Direct ? Single : Multiple) + .mode(connectionMode) + .requiredReplicaSetName(_mongo.getMongoOptions().getRequiredReplicaSetName()) .build(), ServerSettings.builder() .heartbeatFrequency(options.heartbeatFrequencyMS, MILLISECONDS) @@ -324,7 +330,7 @@ private Response innerCall(final DB db, final DBCollection coll, final OutMessag public ServerAddress getAddress() { isTrue("open", !_closed); ClusterDescription clusterDescription = getClusterDescription(); - if (clusterDescription.getConnectionMode() == Single) { + if (connectionMode == Single) { return clusterDescription.getAny().get(0).getAddress(); } if (clusterDescription.getPrimaries().isEmpty()) { @@ -360,16 +366,13 @@ public List getServerAddressList() { public ReplicaSetStatus getReplicaSetStatus() { isTrue("open", !_closed); - ClusterDescription description = getClusterDescription(); - return description.getType() == ReplicaSet && - description.getConnectionMode() == Multiple - ? new ReplicaSetStatus(description) : null; + return getType() == ReplicaSet && connectionMode == Multiple ? new ReplicaSetStatus(getClusterDescription()) : null; } // This call can block if it's not yet known. boolean isMongosConnection() { isTrue("open", !_closed); - return getClusterDescription().getType() == Sharded; + return getType() == Sharded; } public String getConnectPoint(){ @@ -391,8 +394,7 @@ private boolean shouldRetryQuery(ReadPreference readPreference, final DBCollecti if (readPreference.equals(ReadPreference.primary())) { return false; } - ClusterDescription description = getClusterDescription(); - return description.getConnectionMode() == Multiple && description.getType() == ReplicaSet; + return connectionMode == Multiple && getType() == ReplicaSet; } private ClusterDescription getClusterDescription() { @@ -499,7 +501,7 @@ void requestEnsureConnection(){ if ( getPinnedRequestPortForThread() != null ) return; - setPinnedRequestPortForThread(getConnection(new ReadPreferenceServerSelector(ReadPreference.primary()))); + setPinnedRequestPortForThread(getConnection(createServerSelector(ReadPreference.primary()))); } private DBPort getConnection(final ServerSelector serverSelector) { @@ -549,23 +551,37 @@ void setPinnedRequestPortForThread(final DBPort port) { private final ThreadLocal pinnedRequestStatusThreadLocal = new ThreadLocal(); } - private ServerSelector createServerSelector(final ReadPreference readPref) { - return new CompositeServerSelector(Arrays.asList(getPrefixedServerSelector(), - new ReadPreferenceServerSelector(readPref), - new LatencyMinimizingServerSelector(_mongo.getMongoOptions() - .acceptableLatencyDifferenceMS, MILLISECONDS))); - } - - private synchronized ServerSelector getPrefixedServerSelector() { - if (prefixedServerSelector == null) { - ClusterDescription clusterDescription = getClusterDescription(); - if (clusterDescription.getConnectionMode() == Multiple && clusterDescription.getType() == Sharded) { - prefixedServerSelector = new StickyHAShardedClusterServerSelector(); + private ServerSelector createServerSelector(final ReadPreference readPreference) { + if (connectionMode == Multiple) { + List serverSelectorList = new ArrayList(); + if (getType() == Sharded) { + serverSelectorList.add(getMongosHAServerSelector()); + } else if (getType() == ReplicaSet) { + serverSelectorList.add(new ReadPreferenceServerSelector(readPreference)); } else { - prefixedServerSelector = new NoOpServerSelector(); + serverSelectorList.add(new AnyServerSelector()); } + serverSelectorList.add(new LatencyMinimizingServerSelector(_mongo.getMongoOptions().acceptableLatencyDifferenceMS, + MILLISECONDS)); + return new CompositeServerSelector(serverSelectorList); + } else { + return new AnyServerSelector(); + } + } + + private synchronized ClusterType getType() { + if (type == Unknown) { + type = getClusterDescription().getType(); + } + return type; + } + + // There needs to be just one instance of this because it's stateful between requests + private synchronized MongosHAServerSelector getMongosHAServerSelector() { + if (mongosHAServerSelector == null) { + mongosHAServerSelector = new MongosHAServerSelector(); } - return prefixedServerSelector; + return mongosHAServerSelector; } static class PinnedRequestStatus { diff --git a/src/main/com/mongodb/DefaultServer.java b/src/main/com/mongodb/DefaultServer.java index 161eb6e8a04..56f726c1394 100644 --- a/src/main/com/mongodb/DefaultServer.java +++ b/src/main/com/mongodb/DefaultServer.java @@ -108,7 +108,7 @@ public boolean isClosed() { private final class DefaultServerStateListener implements ChangeListener { @Override - public synchronized void stateChanged(final ChangeEvent event) { + public void stateChanged(final ChangeEvent event) { description = event.getNewValue(); for (ChangeListener listener : changeListeners.keySet()) { listener.stateChanged(event); diff --git a/src/main/com/mongodb/DeleteCommandMessage.java b/src/main/com/mongodb/DeleteCommandMessage.java index 2d54cecb1b7..cd90822a9a0 100644 --- a/src/main/com/mongodb/DeleteCommandMessage.java +++ b/src/main/com/mongodb/DeleteCommandMessage.java @@ -51,7 +51,7 @@ protected BaseWriteCommandMessage writeTheWrites(final OutputBuffer buffer, fina writer.writeInt32("limit", remove.isMulti() ? 0 : 1); writer.popMaxDocumentSize(); writer.writeEndDocument(); - if (maximumCommandDocumentSizeExceeded(buffer, commandStartPosition)) { + if (exceedsLimits(buffer.getPosition() - commandStartPosition, i + 1)) { writer.reset(); nextMessage = new DeleteCommandMessage(getWriteNamespace(), getWriteConcern(), deletes.subList(i, deletes.size()), getCommandEncoder(), queryEncoder, getSettings()); diff --git a/src/main/com/mongodb/InsertCommandMessage.java b/src/main/com/mongodb/InsertCommandMessage.java index 43049ee7fac..903518cfc12 100644 --- a/src/main/com/mongodb/InsertCommandMessage.java +++ b/src/main/com/mongodb/InsertCommandMessage.java @@ -43,10 +43,9 @@ protected InsertCommandMessage writeTheWrites(final OutputBuffer buffer, final i for (int i = 0; i < documents.size(); i++) { writer.mark(); writer.encodeDocument(encoder, documents.get(i)); - if (maximumCommandDocumentSizeExceeded(buffer, commandStartPosition)) { + if (exceedsLimits(buffer.getPosition() - commandStartPosition, i + 1)) { writer.reset(); - nextMessage = new InsertCommandMessage(getWriteNamespace(), getWriteConcern(), - documents.subList(i, documents.size()), + nextMessage = new InsertCommandMessage(getWriteNamespace(), getWriteConcern(), documents.subList(i, documents.size()), getCommandEncoder(), encoder, getSettings()); break; } diff --git a/src/main/com/mongodb/LatencyMinimizingServerSelector.java b/src/main/com/mongodb/LatencyMinimizingServerSelector.java index dfad05b7193..76628a9bc5a 100644 --- a/src/main/com/mongodb/LatencyMinimizingServerSelector.java +++ b/src/main/com/mongodb/LatencyMinimizingServerSelector.java @@ -21,14 +21,15 @@ import java.util.Set; import java.util.concurrent.TimeUnit; +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static java.util.concurrent.TimeUnit.NANOSECONDS; + class LatencyMinimizingServerSelector implements ServerSelector { - private final long acceptableLatencyDifference; - private final TimeUnit timeUnit; + private final long acceptableLatencyDifferenceNanos; LatencyMinimizingServerSelector(final long acceptableLatencyDifference, final TimeUnit timeUnit) { - this.acceptableLatencyDifference = acceptableLatencyDifference; - this.timeUnit = timeUnit; + this.acceptableLatencyDifferenceNanos = NANOSECONDS.convert(acceptableLatencyDifference, timeUnit); } @Override @@ -39,8 +40,7 @@ public List choose(final ClusterDescription clusterDescriptio @Override public String toString() { return "LatencyMinimizingServerSelector{" - + "acceptableLatencyDifference=" + acceptableLatencyDifference - + ", timeUnit=" + timeUnit + + "acceptableLatencyDifference=" + MILLISECONDS.convert(acceptableLatencyDifferenceNanos, NANOSECONDS) + " ms" + '}'; } @@ -64,7 +64,7 @@ private List getServersWithAcceptableLatencyDifference(final if (!cur.isOk()) { continue; } - if (cur.getAveragePingTimeNanos() - TimeUnit.NANOSECONDS.convert(acceptableLatencyDifference, timeUnit) <= bestPingTime) { + if (cur.getAveragePingTimeNanos() - acceptableLatencyDifferenceNanos <= bestPingTime) { goodSecondaries.add(cur); } } diff --git a/src/main/com/mongodb/MessageSettings.java b/src/main/com/mongodb/MessageSettings.java index 95c6c67284b..b40061f991d 100644 --- a/src/main/com/mongodb/MessageSettings.java +++ b/src/main/com/mongodb/MessageSettings.java @@ -22,9 +22,11 @@ final class MessageSettings { private static final int DEFAULT_MAX_DOCUMENT_SIZE = 0x1000000; // 16MB private static final int DEFAULT_MAX_MESSAGE_SIZE = 0x2000000; // 32MB + private static final int DEFAULT_MAX_WRITE_BATCH_SIZE = 1000; private final int maxDocumentSize; private final int maxMessageSize; + private final int maxWriteBatchSize; public static Builder builder() { return new Builder(); @@ -33,6 +35,7 @@ public static Builder builder() { static final class Builder { private int maxDocumentSize = DEFAULT_MAX_DOCUMENT_SIZE; private int maxMessageSize = DEFAULT_MAX_MESSAGE_SIZE; + private int maxWriteBatchSize = DEFAULT_MAX_WRITE_BATCH_SIZE; public MessageSettings build() { return new MessageSettings(this); @@ -48,6 +51,12 @@ public Builder maxMessageSize(final int maxMessageSize) { this.maxMessageSize = maxMessageSize; return this; } + + public Builder maxWriteBatchSize(final int maxWriteBatchSize) { + this.maxWriteBatchSize = maxWriteBatchSize; + return this; + } + // CHECKSTYLE:ON } @@ -59,8 +68,13 @@ public int getMaxMessageSize() { return maxMessageSize; } + public int getMaxWriteBatchSize() { + return maxWriteBatchSize; + } + MessageSettings(final Builder builder) { this.maxDocumentSize = builder.maxDocumentSize; this.maxMessageSize = builder.maxMessageSize; + this.maxWriteBatchSize = builder.maxWriteBatchSize; } } \ No newline at end of file diff --git a/src/main/com/mongodb/Mongo.java b/src/main/com/mongodb/Mongo.java index a0526c52483..165f960314a 100644 --- a/src/main/com/mongodb/Mongo.java +++ b/src/main/com/mongodb/Mongo.java @@ -91,7 +91,7 @@ public class Mongo { @Deprecated public static final int MINOR_VERSION = 12; - private static final String FULL_VERSION = "2.12.0-SNAPSHOT"; + private static final String FULL_VERSION = "2.12.0"; static int cleanerIntervalMS; diff --git a/src/main/com/mongodb/MongoClientOptions.java b/src/main/com/mongodb/MongoClientOptions.java index 5c9de009e02..38a9c56a34b 100644 --- a/src/main/com/mongodb/MongoClientOptions.java +++ b/src/main/com/mongodb/MongoClientOptions.java @@ -64,6 +64,7 @@ public static class Builder { private int heartbeatSocketTimeout = Integer.parseInt(System.getProperty("com.mongodb.updaterSocketTimeoutMS", "20000")); private int heartbeatThreadCount; private int acceptableLatencyDifference = Integer.parseInt(System.getProperty("com.mongodb.slaveAcceptableLatencyMS", "15")); + private String requiredReplicaSetName; /** * Sets the heartbeat frequency. @@ -453,6 +454,20 @@ public Builder alwaysUseMBeans(final boolean alwaysUseMBeans) { return this; } + + /** + * Sets the required replica set name for the cluster. + * + * @param requiredReplicaSetName the required replica set name for the replica set. + * @return this + * @see MongoClientOptions#getRequiredReplicaSetName() + * @since 2.12 + */ + public Builder requiredReplicaSetName(final String requiredReplicaSetName) { + this.requiredReplicaSetName = requiredReplicaSetName; + return this; + } + /** * Sets defaults to be what they are in {@code MongoOptions}. * @@ -473,7 +488,7 @@ public Builder legacyDefaults() { public MongoClientOptions build() { return new MongoClientOptions(this); } - } + } /** * Create a new Builder instance. This is a convenience method, equivalent to {@code new MongoClientOptions.Builder()}. @@ -801,6 +816,22 @@ public int getAcceptableLatencyDifference() { return acceptableLatencyDifference; } + /** + * Gets the required replica set name. With this option set, the MongoClient instance will + *

1. Connect in replica set mode, and discover all members of the set based on the given servers + *

+ *

2. Make sure that the set name reported by all members matches the required set name. + *

+ *

3. Refuse to service any requests if any member of the seed list is not part of a replica set with the required name. + *

+ * + * @return the required replica set name + * @since 2.12 + */ + public String getRequiredReplicaSetName() { + return requiredReplicaSetName; + } + @Override public boolean equals(final Object o) { if (this == o) { @@ -887,6 +918,10 @@ public boolean equals(final Object o) { if (!writeConcern.equals(that.writeConcern)) { return false; } + if (requiredReplicaSetName != null ? !requiredReplicaSetName.equals(that.requiredReplicaSetName) + : that.requiredReplicaSetName != null) { + return false; + } return true; } @@ -918,6 +953,7 @@ public int hashCode() { result = 31 * result + heartbeatSocketTimeout; result = 31 * result + heartbeatThreadCount; result = 31 * result + acceptableLatencyDifference; + result = 31 * result + (requiredReplicaSetName != null ? requiredReplicaSetName.hashCode() : 0); return result; } @@ -946,6 +982,7 @@ public String toString() { + ", heartbeatSocketTimeout=" + heartbeatSocketTimeout + ", heartbeatThreadCount=" + heartbeatThreadCount + ", acceptableLatencyDifference=" + acceptableLatencyDifference + + ", requiredReplicaSetName=" + requiredReplicaSetName + '}'; } @@ -975,6 +1012,7 @@ private MongoClientOptions(final Builder builder) { heartbeatSocketTimeout = builder.heartbeatSocketTimeout; heartbeatThreadCount = builder.heartbeatThreadCount; acceptableLatencyDifference = builder.acceptableLatencyDifference; + requiredReplicaSetName = builder.requiredReplicaSetName; } @@ -1003,4 +1041,5 @@ private MongoClientOptions(final Builder builder) { private final int heartbeatSocketTimeout; private final int heartbeatThreadCount; private final int acceptableLatencyDifference; + private final String requiredReplicaSetName; } diff --git a/src/main/com/mongodb/MongoClientURI.java b/src/main/com/mongodb/MongoClientURI.java index e41ae45a20d..19a5795a0f6 100644 --- a/src/main/com/mongodb/MongoClientURI.java +++ b/src/main/com/mongodb/MongoClientURI.java @@ -66,10 +66,13 @@ *
  • {@code ssl=true|false}: Whether to connect using SSL.
    • *
  • {@code connectTimeoutMS=ms}: How long a connection can take to be opened before timing out.
    • *
  • {@code socketTimeoutMS=ms}: How long a send or receive on a socket can take before timing out.
    • + *
  • {@code maxIdleTimeMS=ms}: Maximum idle time of a pooled connection. A connection that exceeds this limit will be closed
    • + *
  • {@code maxLifeTimeMS=ms}: Maximum life time of a pooled connection. A connection that exceeds this limit will be closed
    • * *

    Connection pool configuration:

    *
      *
    • {@code maxPoolSize=n}: The maximum number of connections in the connection pool.
      • + *
    • {@code minPoolSize=n}: The minimum number of connections in the connection pool.
      • *
    • {@code waitQueueMultiple=n} : this multiplier, multiplied with the maxPoolSize setting, gives the maximum number of * threads that may be waiting for a connection to become available from the pool. All further threads will get an * exception right away.
      • @@ -262,14 +265,18 @@ public MongoClientURI(String uri, MongoClientOptions.Builder builder) { static Set allKeys = new HashSet(); static { + generalOptionsKeys.add("minpoolsize"); generalOptionsKeys.add("maxpoolsize"); generalOptionsKeys.add("waitqueuemultiple"); generalOptionsKeys.add("waitqueuetimeoutms"); generalOptionsKeys.add("connecttimeoutms"); + generalOptionsKeys.add("maxidletimems"); + generalOptionsKeys.add("maxlifetimems"); generalOptionsKeys.add("sockettimeoutms"); generalOptionsKeys.add("sockettimeoutms"); generalOptionsKeys.add("autoconnectretry"); generalOptionsKeys.add("ssl"); + generalOptionsKeys.add("replicaset"); readPreferenceKeys.add("slaveok"); readPreferenceKeys.add("readpreference"); @@ -308,6 +315,12 @@ private MongoClientOptions createOptions(Map> optionsMap, M if (key.equals("maxpoolsize")) { builder.connectionsPerHost(Integer.parseInt(value)); + } else if (key.equals("minpoolsize")) { + builder.minConnectionsPerHost(Integer.parseInt(value)); + } else if (key.equals("maxidletimems")) { + builder.maxConnectionIdleTime(Integer.parseInt(value)); + } else if (key.equals("maxlifetimems")) { + builder.maxConnectionLifeTime(Integer.parseInt(value)); } else if (key.equals("waitqueuemultiple")) { builder.threadsAllowedToBlockForConnectionMultiplier(Integer.parseInt(value)); } else if (key.equals("waitqueuetimeoutms")) { @@ -318,6 +331,8 @@ private MongoClientOptions createOptions(Map> optionsMap, M builder.socketTimeout(Integer.parseInt(value)); } else if (key.equals("autoconnectretry")) { builder.autoConnectRetry(_parseBoolean(value)); + } else if (key.equals("replicaset")) { + builder.requiredReplicaSetName(value); } else if (key.equals("ssl")) { if (_parseBoolean(value)) { builder.socketFactory(SSLSocketFactory.getDefault()); diff --git a/src/main/com/mongodb/MongoCursorNotFoundException.java b/src/main/com/mongodb/MongoCursorNotFoundException.java index a28c9aafb6c..d2b78281887 100644 --- a/src/main/com/mongodb/MongoCursorNotFoundException.java +++ b/src/main/com/mongodb/MongoCursorNotFoundException.java @@ -18,6 +18,8 @@ /** * Subclass of {@link MongoException} representing a cursor-not-found exception. + * + * @since 2.12 */ public class MongoCursorNotFoundException extends MongoException { diff --git a/src/main/com/mongodb/MongoOptions.java b/src/main/com/mongodb/MongoOptions.java index 1e287fbbf7b..4be62d71d6b 100644 --- a/src/main/com/mongodb/MongoOptions.java +++ b/src/main/com/mongodb/MongoOptions.java @@ -68,6 +68,7 @@ public MongoOptions(final MongoClientOptions options) { heartbeatReadTimeoutMS = options.getHeartbeatSocketTimeout(); heartbeatThreadCount = options.getHeartbeatThreadCount(); acceptableLatencyDifferenceMS = options.getAcceptableLatencyDifference(); + requiredReplicaSetName = options.getRequiredReplicaSetName(); } public void reset(){ @@ -102,6 +103,7 @@ public void reset(){ heartbeatReadTimeoutMS = Integer.parseInt(System.getProperty("com.mongodb.updaterSocketTimeoutMS", "20000")); heartbeatThreadCount = 0; acceptableLatencyDifferenceMS = Integer.parseInt(System.getProperty("com.mongodb.slaveAcceptableLatencyMS", "15")); + requiredReplicaSetName = null; } public MongoOptions copy() { @@ -137,6 +139,7 @@ public MongoOptions copy() { m.heartbeatReadTimeoutMS = heartbeatReadTimeoutMS; m.heartbeatThreadCount = heartbeatThreadCount; m.acceptableLatencyDifferenceMS = acceptableLatencyDifferenceMS; + m.requiredReplicaSetName = requiredReplicaSetName; return m; } @@ -250,6 +253,10 @@ public boolean equals(final Object o) { if (writeConcern != null ? !writeConcern.equals(options.writeConcern) : options.writeConcern != null) { return false; } + if (requiredReplicaSetName != null ? !requiredReplicaSetName.equals(options.requiredReplicaSetName) + : options.requiredReplicaSetName != null) { + return false; + } return true; } @@ -284,6 +291,7 @@ public int hashCode() { result = 31 * result + heartbeatReadTimeoutMS; result = 31 * result + acceptableLatencyDifferenceMS; result = 31 * result + heartbeatThreadCount; + result = 31 * result + (requiredReplicaSetName != null ? requiredReplicaSetName.hashCode() : 0); return result; } @@ -469,6 +477,8 @@ public int hashCode() { int acceptableLatencyDifferenceMS; int heartbeatThreadCount; + String requiredReplicaSetName; + /** * @return The description for MongoClient instances created with these options */ @@ -808,6 +818,16 @@ public void setAlwaysUseMBeans(final boolean alwaysUseMBeans) { this.alwaysUseMBeans = alwaysUseMBeans; } + /** + * Gets the required replica set name that this client should be connecting to. + * + * @return the required replica set name, or null if none is required + * @since 2.12 + */ + public String getRequiredReplicaSetName() { + return requiredReplicaSetName; + } + @Override public String toString() { return "MongoOptions{" + @@ -833,6 +853,7 @@ public String toString() { ", cursorFinalizerEnabled=" + cursorFinalizerEnabled + ", writeConcern=" + writeConcern + ", alwaysUseMBeans=" + alwaysUseMBeans + + ", requiredReplicaSetName=" + requiredReplicaSetName + '}'; } } diff --git a/src/main/com/mongodb/MongoSocketException.java b/src/main/com/mongodb/MongoSocketException.java index 74a6e7a7371..601b9af65f7 100644 --- a/src/main/com/mongodb/MongoSocketException.java +++ b/src/main/com/mongodb/MongoSocketException.java @@ -19,7 +19,9 @@ import java.io.IOException; /** - * Subclass of {@link MongoException} representing a network-related exception + * Subclass of {@link MongoException} representing a network-related exception. + * + * @since 2.12 */ public class MongoSocketException extends MongoException { diff --git a/src/main/com/mongodb/StickyHAShardedClusterServerSelector.java b/src/main/com/mongodb/MongosHAServerSelector.java similarity index 83% rename from src/main/com/mongodb/StickyHAShardedClusterServerSelector.java rename to src/main/com/mongodb/MongosHAServerSelector.java index b107d830845..af4f46b5c1c 100644 --- a/src/main/com/mongodb/StickyHAShardedClusterServerSelector.java +++ b/src/main/com/mongodb/MongosHAServerSelector.java @@ -22,8 +22,8 @@ import java.util.List; import java.util.Set; -class StickyHAShardedClusterServerSelector implements ServerSelector { - private ServerAddress stickyMongos; +class MongosHAServerSelector implements ServerSelector { + private ServerAddress stickTo; private Set consideredServers = new HashSet(); @Override @@ -36,9 +36,9 @@ public List choose(final ClusterDescription clusterDescriptio Set okServers = getOkServers(clusterDescription); synchronized (this) { - if (!consideredServers.containsAll(okServers) || !okServers.contains(stickyMongos)) { - if (stickyMongos != null && !okServers.contains(stickyMongos)) { - stickyMongos = null; + if (!consideredServers.containsAll(okServers) || !okServers.contains(stickTo)) { + if (stickTo != null && !okServers.contains(stickTo)) { + stickTo = null; consideredServers.clear(); } ServerDescription fastestServer = null; @@ -48,22 +48,22 @@ public List choose(final ClusterDescription clusterDescriptio } } if (fastestServer != null) { - stickyMongos = fastestServer.getAddress(); + stickTo = fastestServer.getAddress(); consideredServers.addAll(okServers); } } - if (stickyMongos == null) { + if (stickTo == null) { return Collections.emptyList(); } else { - return Arrays.asList(clusterDescription.getByServerAddress(stickyMongos)); + return Arrays.asList(clusterDescription.getByServerAddress(stickTo)); } } } @Override public String toString() { - return "StickyHAShardedClusterServerSelector{" - + "stickyMongos=" + stickyMongos + return "MongosHAServerSelector{" + + (stickTo == null ? "" : "stickTo=" + stickTo) + '}'; } diff --git a/src/main/com/mongodb/NoOpServerSelector.java b/src/main/com/mongodb/NoOpServerSelector.java deleted file mode 100644 index 4d5f9fa3f60..00000000000 --- a/src/main/com/mongodb/NoOpServerSelector.java +++ /dev/null @@ -1,31 +0,0 @@ -/* - * Copyright (c) 2008-2014 MongoDB, Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package com.mongodb; - -import java.util.List; - -class NoOpServerSelector implements ServerSelector { - @Override - public List choose(final ClusterDescription clusterDescription) { - return clusterDescription.getAny(); - } - - @Override - public String toString() { - return "NoOpServerSelector{}"; - } -} diff --git a/src/main/com/mongodb/ParallelScanOptions.java b/src/main/com/mongodb/ParallelScanOptions.java index 49270375140..e91874c920c 100644 --- a/src/main/com/mongodb/ParallelScanOptions.java +++ b/src/main/com/mongodb/ParallelScanOptions.java @@ -35,12 +35,16 @@ public static class Builder { /** * Set the requested number of cursors to iterate in parallel. + *

      + * Note: this is the maximum number of cursors the server will return, it may return fewer cursors. + *

      * - * @param numCursors the number of cursors requested, which must be >= 1 + * @param numCursors the number of cursors requested, which must be >= 1 and <= 10000 * @return this */ public Builder numCursors(final int numCursors) { isTrue("numCursors >= 1", numCursors >= 1); + isTrue("numCursors <= 10000", numCursors <= 10000); this.numCursors = numCursors; return this; diff --git a/src/main/com/mongodb/ServerVersion.java b/src/main/com/mongodb/ServerVersion.java index 7f778883cbc..f2774e4237c 100644 --- a/src/main/com/mongodb/ServerVersion.java +++ b/src/main/com/mongodb/ServerVersion.java @@ -17,10 +17,10 @@ package com.mongodb; import java.util.ArrayList; -import java.util.Arrays; import java.util.Collections; import java.util.List; +import static java.util.Arrays.asList; import static org.bson.util.Assertions.isTrue; import static org.bson.util.Assertions.notNull; @@ -34,7 +34,7 @@ class ServerVersion implements Comparable { * Creates a server version which will compare as less than all other valid versions */ public ServerVersion() { - this.versionList = Collections.unmodifiableList(Arrays.asList(0, 0, 0)); + this.versionList = Collections.unmodifiableList(asList(0, 0, 0)); } /** @@ -48,6 +48,16 @@ public ServerVersion(final List versionList) { this.versionList = Collections.unmodifiableList(new ArrayList(versionList)); } + /** + * Constructs a new instance with the given major and minor versions and a patch version of 0. + * + * @param majorVersion the major version + * @param minorVersion the minor version + */ + public ServerVersion(final int majorVersion, final int minorVersion) { + this(asList(majorVersion, minorVersion, 0)); + } + /** * Gets the version list. * diff --git a/src/main/com/mongodb/UnacknowledgedBulkWriteResult.java b/src/main/com/mongodb/UnacknowledgedBulkWriteResult.java index 1f9274629c9..c85ace96297 100644 --- a/src/main/com/mongodb/UnacknowledgedBulkWriteResult.java +++ b/src/main/com/mongodb/UnacknowledgedBulkWriteResult.java @@ -43,6 +43,11 @@ public int getRemovedCount() { throw getUnacknowledgedWriteException(); } + @Override + public boolean isModifiedCountAvailable() { + throw getUnacknowledgedWriteException(); + } + @Override public int getModifiedCount() { throw getUnacknowledgedWriteException(); diff --git a/src/main/com/mongodb/UpdateCommandMessage.java b/src/main/com/mongodb/UpdateCommandMessage.java index bb901d7f415..a147334963d 100644 --- a/src/main/com/mongodb/UpdateCommandMessage.java +++ b/src/main/com/mongodb/UpdateCommandMessage.java @@ -46,11 +46,15 @@ protected UpdateCommandMessage writeTheWrites(final OutputBuffer buffer, final i writer.encodeDocument(getCommandEncoder(), update.getQuery()); writer.writeName("u"); writer.encodeDocument(encoder, update.getUpdateDocument()); - writer.writeBoolean("multi", update.isMulti()); - writer.writeBoolean("upsert", update.isUpsert()); + if (update.isMulti()) { + writer.writeBoolean("multi", update.isMulti()); + } + if (update.isUpsert()) { + writer.writeBoolean("upsert", update.isUpsert()); + } writer.popMaxDocumentSize(); writer.writeEndDocument(); - if (maximumCommandDocumentSizeExceeded(buffer, commandStartPosition)) { + if (exceedsLimits(buffer.getPosition() - commandStartPosition, i + 1)) { writer.reset(); nextMessage = new UpdateCommandMessage(getWriteNamespace(), getWriteConcern(), updates.subList(i, updates.size()), getCommandEncoder(), encoder, getSettings()); diff --git a/src/main/com/mongodb/WriteCommandResultHelper.java b/src/main/com/mongodb/WriteCommandResultHelper.java index 87ba861336d..25f89358305 100644 --- a/src/main/com/mongodb/WriteCommandResultHelper.java +++ b/src/main/com/mongodb/WriteCommandResultHelper.java @@ -20,6 +20,9 @@ import java.util.Collections; import java.util.List; +import static com.mongodb.WriteRequest.Type.REPLACE; +import static com.mongodb.WriteRequest.Type.UPDATE; + final class WriteCommandResultHelper { static boolean hasError(final CommandResult commandResult) { @@ -29,7 +32,7 @@ static boolean hasError(final CommandResult commandResult) { static BulkWriteResult getBulkWriteResult(final WriteRequest.Type type, final CommandResult commandResult) { int count = getCount(commandResult); List upsertedItems = getUpsertedItems(commandResult); - return new AcknowledgedBulkWriteResult(type, count - upsertedItems.size(), getModifiedCount(commandResult), upsertedItems); + return new AcknowledgedBulkWriteResult(type, count - upsertedItems.size(), getModifiedCount(type, commandResult), upsertedItems); } static BulkWriteException getBulkWriteException(final WriteRequest.Type type, final CommandResult commandResult) { @@ -84,8 +87,12 @@ private static int getCount(final CommandResult commandResult) { return commandResult.getInt("n"); } - private static Integer getModifiedCount(final CommandResult commandResult) { - return commandResult.getInt("nModified", 0); + private static Integer getModifiedCount(final WriteRequest.Type type, final CommandResult commandResult) { + Integer modifiedCount = (Integer) commandResult.get("nModified"); + if (modifiedCount == null && !(type == UPDATE || type == REPLACE)) { + modifiedCount = 0; + } + return modifiedCount; } private static DBObject getErrInfo(final DBObject response) { diff --git a/src/main/com/mongodb/gridfs/GridFS.java b/src/main/com/mongodb/gridfs/GridFS.java index 86a9f82ded9..457ebc98bf4 100644 --- a/src/main/com/mongodb/gridfs/GridFS.java +++ b/src/main/com/mongodb/gridfs/GridFS.java @@ -49,7 +49,7 @@ public class GridFS { /** * file's chunk size */ - public static final int DEFAULT_CHUNKSIZE = 256 * 1024; + public static final int DEFAULT_CHUNKSIZE = 255 * 1024; /** * file's max chunk size @@ -296,16 +296,23 @@ public void remove( String filename ){ /** * removes all files matching the given query + * * @param query - * @throws MongoException + * @throws MongoException */ - public void remove( DBObject query ){ - if(query == null) { - throw new IllegalArgumentException("query can not be null"); - } - for ( GridFSDBFile f : find( query ) ){ - f.remove(); + public void remove(DBObject query) { + if (query == null) { + throw new IllegalArgumentException("query can not be null"); + } + // can't remove chunks without files_id thus keep them + List filesIds = new ArrayList(); + for (GridFSDBFile f : find(query)) { + filesIds.add((ObjectId) f.getId()); } + // remove files from bucket + getFilesCollection().remove(query); + // then remove chunks + getChunksCollection().remove(new BasicDBObject("files_id", new BasicDBObject("$in", filesIds))); } diff --git a/src/test/com/mongodb/BulkWriteBatchCombinerSpecification.groovy b/src/test/com/mongodb/BulkWriteBatchCombinerSpecification.groovy index de507f72f01..1c81881f3cb 100644 --- a/src/test/com/mongodb/BulkWriteBatchCombinerSpecification.groovy +++ b/src/test/com/mongodb/BulkWriteBatchCombinerSpecification.groovy @@ -48,6 +48,18 @@ class BulkWriteBatchCombinerSpecification extends Specification { result == new AcknowledgedBulkWriteResult(INSERT, 1, 0, []) } + def 'should handle null modifiedCount'() { + def runResults = new BulkWriteBatchCombiner(new ServerAddress(), ACKNOWLEDGED) + runResults.addResult(new AcknowledgedBulkWriteResult(UPDATE, 1, null, []), new IndexMap.RangeBased().add(0, 0)) + runResults.addResult(new AcknowledgedBulkWriteResult(INSERT, 1, 0, []), new IndexMap.RangeBased().add(0, 0)) + + when: + def result = runResults.getResult() + + then: + result == new AcknowledgedBulkWriteResult(1, 1, 0, null, []) + } + def 'should sort upserts'() { given: def runResults = new BulkWriteBatchCombiner(new ServerAddress(), ACKNOWLEDGED) diff --git a/src/test/com/mongodb/BulkWriteOperationSpecification.groovy b/src/test/com/mongodb/BulkWriteOperationSpecification.groovy index 77dde79d9c0..53b545f860e 100644 --- a/src/test/com/mongodb/BulkWriteOperationSpecification.groovy +++ b/src/test/com/mongodb/BulkWriteOperationSpecification.groovy @@ -1,7 +1,7 @@ /* - * Copyright (c) 2008 - 2013 MongoDB Inc. + * Copyright (c) 2008-2014 MongoDB Inc. * - * Licensed under the Apache License, Version 2.0 (the "License"); + * Licensed under the Apache License, Version 2.0 (the "License") * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * @@ -17,56 +17,105 @@ package com.mongodb import org.bson.types.ObjectId +import spock.lang.Unroll +import static com.mongodb.Fixture.isReplicaSet +import static com.mongodb.Fixture.isServerStartedWithJournalingDisabled +import static com.mongodb.Fixture.isSharded +import static com.mongodb.Fixture.isStandalone +import static com.mongodb.Fixture.serverIsAtLeastVersion import static com.mongodb.WriteRequest.Type.INSERT import static com.mongodb.WriteRequest.Type.REMOVE import static com.mongodb.WriteRequest.Type.REPLACE import static com.mongodb.WriteRequest.Type.UPDATE +import static org.junit.Assume.assumeTrue +@Unroll class BulkWriteOperationSpecification extends FunctionalSpecification { def 'when no document with the same id exists, should insert the document'() { given: - def builder = collection.initializeOrderedBulkOperation(); - builder.insert(new BasicDBObject('_id', 1)) + def operation = initializeBulkOperation(ordered) + operation.insert(new BasicDBObject('_id', 1)) when: - def result = builder.execute() + def result = operation.execute() then: result == new AcknowledgedBulkWriteResult(INSERT, 1, []) - result.getUpserts() == [] collection.findOne() == new BasicDBObject('_id', 1) + + where: + ordered << [true, false] + } + + def 'when a document contains a key with an illegal character, inserting it should throw IllegalArgumentException'() { + given: + def operation = initializeBulkOperation(ordered) + operation.insert(new BasicDBObject('$set', 1)) + + when: + operation.execute() + + then: + thrown(IllegalArgumentException) + + where: + ordered << [true, false] } def 'when a document with the same id exists, should throw an exception'() { given: def document = new BasicDBObject('_id', 1) collection.insert(document) - def builder = collection.initializeOrderedBulkOperation(); - builder.insert(document) + def operation = initializeBulkOperation(ordered) + operation.insert(document) when: - builder.execute() + operation.execute() then: def ex = thrown(BulkWriteException) ex.getWriteErrors().get(0).code == 11000 + + where: + ordered << [true, false] + } + + def 'when a document with no _id is inserted, the _id should be generated by the driver'() { + given: + def operation = initializeBulkOperation(ordered) + def document = new BasicDBObject() + operation.insert(document) + + when: + def result = operation.execute() + + then: + result == new AcknowledgedBulkWriteResult(INSERT, 1, []) + document._id instanceof ObjectId + collection.findOne() == document + + where: + ordered << [true, false] } def 'when documents match the query, a remove of one should remove one of them'() { given: collection.insert(new BasicDBObject('x', true)) collection.insert(new BasicDBObject('x', true)) - def builder = collection.initializeOrderedBulkOperation() - builder.find(new BasicDBObject('x', true)).removeOne() + def operation = initializeBulkOperation(ordered) + operation.find(new BasicDBObject('x', true)).removeOne() when: - def result = builder.execute() + def result = operation.execute() then: result == new AcknowledgedBulkWriteResult(REMOVE, 1, []) collection.count() == 1 + + where: + ordered << [true, false] } def 'when documents match the query, a remove should remove all of them'() { @@ -74,121 +123,279 @@ class BulkWriteOperationSpecification extends FunctionalSpecification { collection.insert(new BasicDBObject('x', true)) collection.insert(new BasicDBObject('x', true)) collection.insert(new BasicDBObject('x', false)) - def builder = collection.initializeOrderedBulkOperation() - builder.find(new BasicDBObject('x', true)).remove() + def operation = initializeBulkOperation(ordered) + operation.find(new BasicDBObject('x', true)).remove() when: - def result = builder.execute() + def result = operation.execute() then: result == new AcknowledgedBulkWriteResult(REMOVE, 2, []) - collection.count() == 1 + collection.count(new BasicDBObject('x', false)) == 1 + + where: + ordered << [true, false] } - def 'when a document matches the query, an update of one should update that document'() { + def 'when an update document contains a non $-prefixed key, update should throw IllegalArgumentException'() { + given: + def operation = initializeBulkOperation(ordered) + operation.find(new BasicDBObject()).update(new BasicDBObject('$set', new BasicDBObject('x', 1)).append('y', 2)) + + when: + operation.execute() + + then: + thrown(IllegalArgumentException) + + where: + ordered << [true, false] + } + + def 'when an update document contains a non $-prefixed key, updateOne should throw IllegalArgumentException'() { + given: + def operation = initializeBulkOperation(ordered) + operation.find(new BasicDBObject()).updateOne(new BasicDBObject('$set', new BasicDBObject('x', 1)).append('y', 2)) + + when: + operation.execute() + + then: + thrown(IllegalArgumentException) + + where: + ordered << [true, false] + } + + def 'when multiple document match the query, updateOne should update only one of them'() { given: collection.insert(new BasicDBObject('x', true)) collection.insert(new BasicDBObject('x', true)) - def builder = collection.initializeOrderedBulkOperation() - builder.find(new BasicDBObject('x', true)).updateOne(new BasicDBObject('$set', new BasicDBObject('y', 1))) + def operation = initializeBulkOperation(ordered) + operation.find(new BasicDBObject('x', true)).updateOne(new BasicDBObject('$set', new BasicDBObject('y', 1))) when: - def result = builder.execute() + def result = operation.execute() then: - result == new AcknowledgedBulkWriteResult(UPDATE, 1, 1, []) - collection.count(new BasicDBObject('y', 1)) == 1 + result == new AcknowledgedBulkWriteResult(UPDATE, 1, expectedModifiedCount(1), []) + collection.find(new BasicDBObject('y', 1), new BasicDBObject('x', 1).append('_id', 0)).toArray() == [new BasicDBObject('x', true)] + + where: + ordered << [true, false] } - def 'when documents match the query, an update should update all of them'() { + def 'when documents match the query, an update should update all of them'() { given: collection.insert(new BasicDBObject('x', true)) collection.insert(new BasicDBObject('x', true)) collection.insert(new BasicDBObject('x', false)) - def builder = collection.initializeOrderedBulkOperation() - builder.find(new BasicDBObject('x', true)).update(new BasicDBObject('$set', new BasicDBObject('y', 1))) + def operation = initializeBulkOperation(ordered) + operation.find(new BasicDBObject('x', true)).update(new BasicDBObject('$set', new BasicDBObject('y', 1))) when: - def result = builder.execute() + def result = operation.execute() then: - result == new AcknowledgedBulkWriteResult(UPDATE, 2, 2, []) + result == new AcknowledgedBulkWriteResult(UPDATE, 2, expectedModifiedCount(2), []) collection.count(new BasicDBObject('y', 1)) == 2 + + where: + ordered << [true, false] + } + + def 'when no document matches the query, updateOne with upsert should insert a document'() { + given: + def id = new ObjectId() + def operation = initializeBulkOperation(ordered) + operation.find(new BasicDBObject('_id', id)).upsert().updateOne(new BasicDBObject('$set', new BasicDBObject('x', 2))) + + when: + def result = operation.execute() + + then: + result == new AcknowledgedBulkWriteResult(UPDATE, 0, expectedModifiedCount(0), [new BulkWriteUpsert(0, id)]) + collection.findOne() == new BasicDBObject('_id', id).append('x', 2) + + where: + ordered << [true, false] } - def 'when no document matches the query, an update with upsert should insert a document'() { + def 'when no document matches the query, update with upsert should insert a document'() { given: def id = new ObjectId() - def builder = collection.initializeOrderedBulkOperation() - builder.find(new BasicDBObject('_id', id)).upsert().updateOne(new BasicDBObject('$set', new BasicDBObject('x', 2))) + def operation = initializeBulkOperation(ordered) + operation.find(new BasicDBObject('_id', id)).upsert().update(new BasicDBObject('$set', new BasicDBObject('x', 2))) when: - def result = builder.execute() + def result = operation.execute() then: - result == new AcknowledgedBulkWriteResult(UPDATE, 0, [new BulkWriteUpsert(0, id)]) + result == new AcknowledgedBulkWriteResult(UPDATE, 0, expectedModifiedCount(0), [new BulkWriteUpsert(0, id)]) collection.findOne() == new BasicDBObject('_id', id).append('x', 2) + + where: + ordered << [true, false] + } + + def 'when documents matches the query, update with upsert should update all of them'() { + given: + collection.insert(new BasicDBObject('x', true)) + collection.insert(new BasicDBObject('x', true)) + collection.insert(new BasicDBObject('x', false)) + + def operation = initializeBulkOperation(ordered) + operation.find(new BasicDBObject('x', true)).upsert().update(new BasicDBObject('$set', new BasicDBObject('y', 1))) + + when: + def result = operation.execute() + + then: + result == new AcknowledgedBulkWriteResult(UPDATE, 2, expectedModifiedCount(2), []) + collection.count(new BasicDBObject('y', 1)) == 2 + + where: + ordered << [true, false] + } + + def 'when a document contains a key with an illegal character, replacing a document with it should throw IllegalArgumentException'() { + given: + def id = new ObjectId() + def operation = initializeBulkOperation(ordered) + def query = new BasicDBObject('_id', id) + operation.find(query).upsert().replaceOne(new BasicDBObject('$set', new BasicDBObject('x', 1))) + + when: + operation.execute() + + then: + thrown(IllegalArgumentException) + + where: + ordered << [true, false] } def 'when no document matches the query, a replace with upsert should insert a document'() { given: def id = new ObjectId() - def builder = collection.initializeOrderedBulkOperation() + def operation = initializeBulkOperation(ordered) def query = new BasicDBObject('_id', id) - builder.find(query).upsert().replaceOne(new BasicDBObject('_id', id).append('x', 2)) + operation.find(query).upsert().replaceOne(new BasicDBObject('_id', id).append('x', 2)) when: - def result = builder.execute() + def result = operation.execute() then: - result == new AcknowledgedBulkWriteResult(UPDATE, 0, [new BulkWriteUpsert(0, id)]) + result == new AcknowledgedBulkWriteResult(UPDATE, 0, expectedModifiedCount(0), [new BulkWriteUpsert(0, id)]) collection.findOne() == new BasicDBObject('_id', id).append('x', 2) + + where: + ordered << [true, false] } - def 'when a document matches the query, an update with upsert should update that document'() { + def 'when multiple documents match the query, replaceOne should replace one of them'() { + given: + collection.insert(new BasicDBObject('x', true)) + collection.insert(new BasicDBObject('x', true)) + + def operation = initializeBulkOperation(ordered) + def replacement = new BasicDBObject('y', 1).append('x', false) + operation.find(new BasicDBObject('x', true)).replaceOne(replacement) + + when: + def result = operation.execute() + + then: + result == new AcknowledgedBulkWriteResult(UPDATE, 1, expectedModifiedCount(1), []) + collection.find(new BasicDBObject('x', false), new BasicDBObject('_id', 0)).toArray() == [replacement] + + where: + ordered << [true, false] + } + + def 'when a document matches the query, updateOne with upsert should update that document'() { given: def id = new ObjectId() collection.insert(new BasicDBObject('_id', id)) - def builder = collection.initializeOrderedBulkOperation() - builder.find(new BasicDBObject('_id', id)).upsert().updateOne(new BasicDBObject('$set', new BasicDBObject('x', 2))) + def operation = initializeBulkOperation(ordered) + operation.find(new BasicDBObject('_id', id)).upsert().updateOne(new BasicDBObject('$set', new BasicDBObject('x', 2))) when: - def result = builder.execute() + def result = operation.execute() then: - result == new AcknowledgedBulkWriteResult(UPDATE, 1, 1, []) + result == new AcknowledgedBulkWriteResult(UPDATE, 1, expectedModifiedCount(1), []) collection.findOne() == new BasicDBObject('_id', id).append('x', 2) + + where: + ordered << [true, false] } def 'when a document matches the query, a replace with upsert should update that document'() { given: collection.insert(new BasicDBObject('_id', 1)) - def builder = collection.initializeOrderedBulkOperation() - builder.find(new BasicDBObject('_id', 1)).upsert().replaceOne(new BasicDBObject('_id', 1).append('x', 2)) + def operation = initializeBulkOperation(ordered) + operation.find(new BasicDBObject('_id', 1)).upsert().replaceOne(new BasicDBObject('_id', 1).append('x', 2)) when: - def result = builder.execute() + def result = operation.execute() then: - result == new AcknowledgedBulkWriteResult(REPLACE, 1, 1, []) + result == new AcknowledgedBulkWriteResult(REPLACE, 1, expectedModifiedCount(1), []) collection.findOne() == new BasicDBObject('_id', 1).append('x', 2) + + where: + ordered << [true, false] + } + + def 'when a replacement document is 16MB, the document is still replaced'() { + given: + collection.insert(new BasicDBObject('_id', 1)) + + def operation = collection.initializeOrderedBulkOperation() + operation.find(new BasicDBObject('_id', 1)).upsert().replaceOne(new BasicDBObject('_id', 1) + .append('x', new byte[1024 * 1024 * 16 - 30])) + + when: + def result = operation.execute() + + then: + result == new AcknowledgedBulkWriteResult(REPLACE, 1, expectedModifiedCount(1), []) + collection.count() == 1 + } + + def 'when two update documents together exceed 16MB, the documents are still updated'() { + given: + def operation = collection.initializeOrderedBulkOperation() + operation.find(new BasicDBObject('_id', 1)) + .update(new BasicDBObject('$set', new BasicDBObject('x', new byte[1024 * 1024 * 8]))); + operation.find(new BasicDBObject('_id', 2)) + .update(new BasicDBObject('$set', new BasicDBObject('x', new byte[1024 * 1024 * 8]))); + + when: + def result = operation.execute() + + then: + result == new AcknowledgedBulkWriteResult(UPDATE, 0, expectedModifiedCount(0), []) + collection.count() == 0 } + def 'should handle multi-length runs of ordered insert, update, replace, and remove'() { given: collection.insert(getTestInserts()) - def builder = collection.initializeOrderedBulkOperation() - addWritesToBuilder(builder) + def operation = collection.initializeOrderedBulkOperation() + addWritesToOperation(operation) when: - def result = builder.execute() + def result = operation.execute() then: - result == new AcknowledgedBulkWriteResult(2, 4, 2, 4, []) + result == new AcknowledgedBulkWriteResult(2, 4, 2, expectedModifiedCount(4), []) collection.findOne(new BasicDBObject('_id', 1)) == new BasicDBObject('_id', 1).append('x', 2) collection.findOne(new BasicDBObject('_id', 2)) == new BasicDBObject('_id', 2).append('x', 3) @@ -205,11 +412,11 @@ class BulkWriteOperationSpecification extends FunctionalSpecification { collection.getDB().requestStart() collection.insert(getTestInserts()) - def builder = collection.initializeOrderedBulkOperation() - addWritesToBuilder(builder) + def operation = initializeBulkOperation(ordered) + addWritesToOperation(operation) when: - def result = builder.execute(WriteConcern.UNACKNOWLEDGED) + def result = operation.execute(WriteConcern.UNACKNOWLEDGED) collection.insert(new BasicDBObject('_id', 9)) then: @@ -225,36 +432,42 @@ class BulkWriteOperationSpecification extends FunctionalSpecification { cleanup: collection.getDB().requestDone() + + where: + ordered << [true, false] } def 'error details should have correct index on ordered write failure'() { given: - def builder = collection.initializeOrderedBulkOperation() - builder.insert(new BasicDBObject('_id', 1)) - builder.find(new BasicDBObject('_id', 1)).updateOne(new BasicDBObject('$set', new BasicDBObject('x', 3))) - builder.insert(new BasicDBObject('_id', 1)) + def operation = initializeBulkOperation(ordered) + operation.insert(new BasicDBObject('_id', 1)) + operation.find(new BasicDBObject('_id', 1)).updateOne(new BasicDBObject('$set', new BasicDBObject('x', 3))) + operation.insert(new BasicDBObject('_id', 1)) when: - builder.execute() + operation.execute() then: def ex = thrown(BulkWriteException) ex.writeErrors.size() == 1 ex.writeErrors[0].index == 2 ex.writeErrors[0].code == 11000 + + where: + ordered << [true, false] } def 'should handle multi-length runs of unordered insert, update, replace, and remove'() { given: - collection.insert(getTestInserts()); - def builder = collection.initializeUnorderedBulkOperation() - addWritesToBuilder(builder) + collection.insert(getTestInserts()) + def operation = collection.initializeUnorderedBulkOperation() + addWritesToOperation(operation) when: - def result = builder.execute() + def result = operation.execute() then: - result == new AcknowledgedBulkWriteResult(2, 4, 2, 4, []) + result == new AcknowledgedBulkWriteResult(2, 4, 2, expectedModifiedCount(4), []) collection.findOne(new BasicDBObject('_id', 1)) == new BasicDBObject('_id', 1).append('x', 2) collection.findOne(new BasicDBObject('_id', 2)) == new BasicDBObject('_id', 2).append('x', 3) @@ -266,17 +479,110 @@ class BulkWriteOperationSpecification extends FunctionalSpecification { collection.findOne(new BasicDBObject('_id', 8)) == new BasicDBObject('_id', 8) } + def 'should split when the number of writes is larger than the match write batch size'() { + given: + def operation = initializeBulkOperation(ordered) + (0..2000).each { + operation.insert(new BasicDBObject()) + } + + when: + operation.execute() + + then: + collection.find().count() == 2001 + + where: + ordered << [true, false] + } + + def 'should split when the message size would exceed the max command message size'() { + given: + def operation = collection.initializeUnorderedBulkOperation() + (0..5).each { + operation.insert(new BasicDBObject('binary', new byte[1024 * 1024 * 4])) + } + + when: + operation.execute() + + then: + collection.count() == 6 + } + + def 'should throw correct BulkWriteException when the message size would exceed the max command message size'(boolean ordered) { + given: + def operation = initializeBulkOperation(ordered) + (0..5).each { + operation.insert(new BasicDBObject('_id', it).append('binary', new byte[1024 * 1024 * 4])) + } + operation.insert(new BasicDBObject('_id', 0)) // duplicate key + operation.insert(new BasicDBObject('_id', 6)) + + when: + operation.execute() + + then: + def ex = thrown(BulkWriteException) + ex.writeErrors.size() == 1 + ex.writeErrors[0].index == 6 + ex.writeResult == new AcknowledgedBulkWriteResult(INSERT, ordered ? 6 : 7, []) // for ordered, last doc will not be inserted + + where: + ordered << [true, false] + } + + def 'should insert all documents when the number of inserts is larger than the match write batch size '(boolean ordered) { + given: + def operation = initializeBulkOperation(ordered) + (0..1001).each { + operation.insert(new BasicDBObject('_id', it)) + } + + when: + operation.execute() + + then: + collection.count == 1002 + + where: + ordered << [true, false] + } + + def 'should throw correct BulkWriteException when the number of writes is larger than the match write batch size '(boolean ordered) { + given: + def operation = initializeBulkOperation(ordered) + (0..999).each { + operation.insert(new BasicDBObject('_id', it)) + } + + operation.insert(new BasicDBObject('_id', 0)) // duplicate key + operation.insert(new BasicDBObject('_id', 1000)) + + when: + operation.execute() + + then: + def ex = thrown(BulkWriteException) + ex.writeErrors.size() == 1 + ex.writeErrors[0].index == 1000 + ex.writeResult == new AcknowledgedBulkWriteResult(INSERT, ordered ? 1000 : 1001, []) // for ordered, last doc will not be inserted + + where: + ordered << [true, false] + } + def 'error details should have correct index on unordered write failure'() { given: collection.insert(getTestInserts()) - def builder = collection.initializeUnorderedBulkOperation() - builder.insert(new BasicDBObject('_id', 1)) - builder.find(new BasicDBObject('_id', 2)).updateOne(new BasicDBObject('$set', new BasicDBObject('x', 3))) - builder.insert(new BasicDBObject('_id', 3)) + def operation = collection.initializeUnorderedBulkOperation() + operation.insert(new BasicDBObject('_id', 1)) + operation.find(new BasicDBObject('_id', 2)).updateOne(new BasicDBObject('$set', new BasicDBObject('x', 3))) + operation.insert(new BasicDBObject('_id', 3)) when: - builder.execute() + operation.execute() then: def ex = thrown(BulkWriteException) @@ -287,103 +593,207 @@ class BulkWriteOperationSpecification extends FunctionalSpecification { ex.writeErrors[1].code == 11000 } + def 'when there is a duplicate key error and a write concern error, both should be reported'() { + assumeTrue(isReplicaSet()) + + given: + collection.insert(getTestInserts()) + + def operation = initializeBulkOperation(ordered) + operation.insert(new BasicDBObject('_id', 7)) + operation.insert(new BasicDBObject('_id', 1)) // duplicate key + + when: + operation.execute(new WriteConcern(4, 1)) // This is assuming that it won't be able to replicate to 4 servers in 1 ms + + then: + def ex = thrown(BulkWriteException) + ex.writeErrors.size() == 1 + ex.writeErrors[0].index == 1 + ex.writeErrors[0].code == 11000 + ex.writeConcernError != null + + where: + ordered << [false] + } + + def 'when w > 1 write concern is used on a standalone server with write commands, CommandFailureException is thrown'() { + assumeTrue(isStandalone() && serverIsAtLeastVersion(2.6)) + + given: + def operation = collection.initializeUnorderedBulkOperation() + operation.insert(new BasicDBObject('_id', 1)) + + when: + operation.execute(new WriteConcern(2, 1)) + + then: + thrown(CommandFailureException) + + where: + ordered << [true, false] + } + + def 'when w > 1 write concern is used on a standalone server without write commands, BulkWriteException is thrown'() { + assumeTrue(isStandalone() && !serverIsAtLeastVersion(2.6)) + + given: + def operation = collection.initializeUnorderedBulkOperation() + operation.insert(new BasicDBObject('_id', 1)) + operation.insert(new BasicDBObject('_id', 2)) + + when: + operation.execute(new WriteConcern(2, 1)) + + then: + def e = thrown(BulkWriteException) + e.writeResult == new AcknowledgedBulkWriteResult(INSERT, 2, []) + e.writeConcernError != null + e.writeConcernError.getDetails().containsField('wnote') + + where: + ordered << [true, false] + } + + def 'when j write concern is used on a server without journaling or write commands, BulkWriteException is thrown'() { + assumeTrue(!isSharded() && isServerStartedWithJournalingDisabled() && !serverIsAtLeastVersion(2.6)) + + given: + def operation = collection.initializeUnorderedBulkOperation() + operation.insert(new BasicDBObject('_id', 1)) + operation.insert(new BasicDBObject('_id', 2)) + + when: + operation.execute(WriteConcern.JOURNALED) + + then: + def e = thrown(BulkWriteException) + e.writeResult == new AcknowledgedBulkWriteResult(INSERT, 2, []) + e.writeConcernError != null + e.writeConcernError.getDetails().containsField('jnote') + + where: + ordered << [true, false] + } + def 'execute should throw IllegalStateException when already executed'() { given: - def builder = collection.initializeOrderedBulkOperation(); - builder.insert(new BasicDBObject('_id', 1)) - builder.execute() + def operation = initializeBulkOperation(ordered) + operation.insert(new BasicDBObject('_id', 1)) + operation.execute() when: - builder.execute() + operation.execute() then: thrown(IllegalStateException) + + where: + ordered << [true, false] } def 'execute with write concern should throw IllegalStateException when already executed'() { given: - def builder = collection.initializeOrderedBulkOperation(); - builder.insert(new BasicDBObject('_id', 1)) - builder.execute() + def operation = initializeBulkOperation(ordered) + operation.insert(new BasicDBObject('_id', 1)) + operation.execute() when: - builder.execute(WriteConcern.ACKNOWLEDGED) + operation.execute(WriteConcern.ACKNOWLEDGED) then: thrown(IllegalStateException) + + where: + ordered << [true, false] } def 'insert should throw IllegalStateException when already executed'() { given: - def builder = collection.initializeOrderedBulkOperation(); - builder.insert(new BasicDBObject('_id', 1)) - builder.execute() + def operation = initializeBulkOperation(ordered) + operation.insert(new BasicDBObject('_id', 1)) + operation.execute() when: - builder.insert(new BasicDBObject()) + operation.insert(new BasicDBObject()) then: thrown(IllegalStateException) + + where: + ordered << [true, false] } def 'find should throw IllegalStateException when already executed'() { given: - def builder = collection.initializeOrderedBulkOperation(); - builder.insert(new BasicDBObject('_id', 1)) - builder.execute() + def operation = initializeBulkOperation(ordered) + operation.insert(new BasicDBObject('_id', 1)) + operation.execute() when: - builder.find(new BasicDBObject()); + operation.find(new BasicDBObject()) then: thrown(IllegalStateException) + + where: + ordered << [true, false] } // just need to check one case here, since the others are checked above def 'should throw IllegalStateException when already executed with write concern'() { given: - def builder = collection.initializeOrderedBulkOperation(); - builder.insert(new BasicDBObject('_id', 1)) - builder.execute(WriteConcern.ACKNOWLEDGED) + def operation = initializeBulkOperation(ordered) + operation.insert(new BasicDBObject('_id', 1)) + operation.execute(WriteConcern.ACKNOWLEDGED) when: - builder.execute() + operation.execute() then: thrown(IllegalStateException) + + where: + ordered << [true, false] } def 'should throw IllegalStateException when executing an empty bulk operation'() { given: - def builder = collection.initializeOrderedBulkOperation(); + def operation = initializeBulkOperation(ordered) when: - builder.execute() + operation.execute() then: thrown(IllegalStateException) + + where: + ordered << [true, false] } def 'should throw IllegalStateException when executing an empty bulk operation with a write concern'() { given: - def builder = collection.initializeOrderedBulkOperation(); + def operation = initializeBulkOperation(ordered) when: - builder.execute(WriteConcern.ACKNOWLEDGED) + operation.execute(WriteConcern.ACKNOWLEDGED) then: thrown(IllegalStateException) + + where: + ordered << [true, false] } - private static void addWritesToBuilder(BulkWriteOperation builder) { - builder.find(new BasicDBObject('_id', 1)).updateOne(new BasicDBObject('$set', new BasicDBObject('x', 2))) - builder.find(new BasicDBObject('_id', 2)).updateOne(new BasicDBObject('$set', new BasicDBObject('x', 3))) - builder.find(new BasicDBObject('_id', 3)).removeOne() - builder.find(new BasicDBObject('_id', 4)).removeOne() - builder.find(new BasicDBObject('_id', 5)).replaceOne(new BasicDBObject('_id', 5).append('x', 4)) - builder.find(new BasicDBObject('_id', 6)).replaceOne(new BasicDBObject('_id', 6).append('x', 5)) - builder.insert(new BasicDBObject('_id', 7)) - builder.insert(new BasicDBObject('_id', 8)) + private static void addWritesToOperation(BulkWriteOperation operation) { + operation.find(new BasicDBObject('_id', 1)).updateOne(new BasicDBObject('$set', new BasicDBObject('x', 2))) + operation.find(new BasicDBObject('_id', 2)).updateOne(new BasicDBObject('$set', new BasicDBObject('x', 3))) + operation.find(new BasicDBObject('_id', 3)).removeOne() + operation.find(new BasicDBObject('_id', 4)).removeOne() + operation.find(new BasicDBObject('_id', 5)).replaceOne(new BasicDBObject('_id', 5).append('x', 4)) + operation.find(new BasicDBObject('_id', 6)).replaceOne(new BasicDBObject('_id', 6).append('x', 5)) + operation.insert(new BasicDBObject('_id', 7)) + operation.insert(new BasicDBObject('_id', 8)) } private static List getTestInserts() { @@ -396,4 +806,11 @@ class BulkWriteOperationSpecification extends FunctionalSpecification { ] } + private BulkWriteOperation initializeBulkOperation(boolean ordered) { + ordered ? collection.initializeOrderedBulkOperation() : collection.initializeUnorderedBulkOperation() + } + + private static Integer expectedModifiedCount(final int expectedCountForServersThatSupportIt) { + (serverIsAtLeastVersion(2.5)) ? expectedCountForServersThatSupportIt : null + } } \ No newline at end of file diff --git a/src/test/com/mongodb/DBCollectionTest.java b/src/test/com/mongodb/DBCollectionTest.java index 6626304c7f9..717f4df4fe7 100644 --- a/src/test/com/mongodb/DBCollectionTest.java +++ b/src/test/com/mongodb/DBCollectionTest.java @@ -21,6 +21,7 @@ import org.junit.Test; import java.net.UnknownHostException; +import java.util.ArrayList; import java.util.Arrays; import java.util.HashSet; import java.util.List; @@ -49,6 +50,17 @@ public void testMultiInsert() { c.insert(inserted1,inserted2); } + @Test + public void testLargeMultiInsert() { + List documents = new ArrayList(); + for (int i = 0; i < 1001; i++) { + documents.add(new BasicDBObject()); + } + + collection.insert(documents); + assertEquals(1001, collection.count()); + } + @Test public void testCappedCollection() { String collectionName = "testCapped"; @@ -181,6 +193,12 @@ public void testDropIndex(){ c.dropIndexes(); assertEquals( 1 , c.getIndexInfo().size() ); + c.createIndex( new BasicDBObject( "x" , 1 ) ); + assertEquals( 2 , c.getIndexInfo().size() ); + + c.dropIndexes(); + assertEquals( 1 , c.getIndexInfo().size() ); + c.ensureIndex( new BasicDBObject( "x" , 1 ) ); assertEquals( 2 , c.getIndexInfo().size() ); @@ -190,6 +208,18 @@ public void testDropIndex(){ c.dropIndex( new BasicDBObject( "x" , 1 ) ); assertEquals( 2 , c.getIndexInfo().size() ); + c.dropIndexes(); + assertEquals( 1 , c.getIndexInfo().size() ); + + c.createIndex( new BasicDBObject( "x" , 1 ) ); + assertEquals( 2 , c.getIndexInfo().size() ); + + c.createIndex( new BasicDBObject( "y" , 1 ) ); + assertEquals( 3 , c.getIndexInfo().size() ); + + c.dropIndex( new BasicDBObject( "x" , 1 ) ); + assertEquals( 2 , c.getIndexInfo().size() ); + c.dropIndexes(); } @Test @@ -246,7 +276,15 @@ public void testEnsureIndex(){ c.ensureIndex( new BasicDBObject( "x" , 1 ) , new BasicDBObject( "unique" , true ) ); assertEquals( 2 , c.getIndexInfo().size() ); - assertEquals( Boolean.TRUE , c.getIndexInfo().get(1).get( "unique" ) ); + DBObject indexInfo = c.getIndexInfo().get(1); + assertEquals("x_1", indexInfo.get("name")); + + c.drop(); + + c.createIndex(new BasicDBObject("x", 1), new BasicDBObject("unique", true)); + indexInfo = c.getIndexInfo().get(1); + assertEquals( Boolean.TRUE , indexInfo.get("unique") ); + assertEquals("x_1", indexInfo.get("name")); } @Test @@ -259,22 +297,47 @@ public void testEnsureNestedIndex(){ assertEquals( 1 , c.getIndexInfo().size() ); c.ensureIndex( new BasicDBObject("x.y", 1), "nestedIdx1", false); assertEquals( 2 , c.getIndexInfo().size() ); + assertEquals( "nestedIdx1" , c.getIndexInfo().get(1).get("name") ); + + c.drop(); + c.createIndex(new BasicDBObject("x.y", 1), new BasicDBObject("name", "nestedIdx1").append("unique", false)); + assertEquals(2, c.getIndexInfo().size()); + assertEquals( "nestedIdx1" , c.getIndexInfo().get(1).get("name") ); } - @Test(expected = MongoException.DuplicateKey.class) - public void testIndexExceptions(){ + @Test(expected = DuplicateKeyException.class) + public void testEnsureIndexExceptions(){ collection.insert(new BasicDBObject("x", 1)); collection.insert(new BasicDBObject("x", 1)); collection.ensureIndex(new BasicDBObject("y", 1)); collection.resetIndexCache(); - collection.ensureIndex(new BasicDBObject("y", 1)); // make sure this doesn't throw + try { + collection.ensureIndex(new BasicDBObject("y", 1)); // make sure this doesn't throw + } catch (Exception e) { + fail("Trying to create an existing index should not fail."); + } collection.resetIndexCache(); collection.ensureIndex(new BasicDBObject("x", 1), new BasicDBObject("unique", true)); } + @Test(expected = DuplicateKeyException.class) + public void testCreateIndexExceptions(){ + collection.insert(new BasicDBObject("x", 1)); + collection.insert(new BasicDBObject("x", 1)); + + collection.createIndex(new BasicDBObject("y", 1)); + try { + collection.createIndex(new BasicDBObject("y", 1)); // make sure this doesn't throw + } catch (Exception e) { + fail("Trying to create an existing index should not fail."); + } + + collection.createIndex(new BasicDBObject("x", 1), new BasicDBObject("unique", true)); + } + @Test public void testMultiInsertNoContinue() { collection.setWriteConcern(WriteConcern.NORMAL); @@ -499,8 +562,9 @@ public void testParallelScan() throws UnknownHostException { collection.insert(new BasicDBObject("_id", i)); } - List cursors = collection.parallelScan(ParallelScanOptions.builder().numCursors(3).batchSize(1000).build()); - assertEquals(3, cursors.size()); + int numCursors = 10; + List cursors = collection.parallelScan(ParallelScanOptions.builder().numCursors(numCursors).batchSize(1000).build()); + assertTrue(cursors.size() <= numCursors); for (Cursor cursor : cursors) { while (cursor.hasNext()) { diff --git a/src/test/com/mongodb/DBCursorTest.java b/src/test/com/mongodb/DBCursorTest.java index ea099ed3136..0facb58a51f 100644 --- a/src/test/com/mongodb/DBCursorTest.java +++ b/src/test/com/mongodb/DBCursorTest.java @@ -499,6 +499,32 @@ public void testMaxTimeForIterator() { } } + @Test + public void testMaxTimeDuringGetMore() { + assumeFalse(isSharded(getMongoClient())); + checkServerVersion(2.5); + for (int i=0; i < 20; i++) { + collection.insert(new BasicDBObject("x", 1)); + } + + DBCursor cursor = new DBCursor(collection, new BasicDBObject("x", 1), new BasicDBObject(), ReadPreference.primary()); + cursor.batchSize(10); + cursor.maxTime(1, SECONDS); + cursor.next(); + + enableMaxTimeFailPoint(); + try { + while(cursor.hasNext()) { + cursor.next(); + } + fail("Show have thrown"); + } catch (MongoExecutionTimeoutException e) { + assertEquals(50, e.getCode()); + } finally { + disableMaxTimeFailPoint(); + } + } + @Test public void testMaxTimeForIterable() { assumeFalse(isSharded(getMongoClient())); diff --git a/src/test/com/mongodb/DBTest.java b/src/test/com/mongodb/DBTest.java index ff8f710bd00..e388ba7e0de 100644 --- a/src/test/com/mongodb/DBTest.java +++ b/src/test/com/mongodb/DBTest.java @@ -25,6 +25,7 @@ import java.net.UnknownHostException; import java.util.Arrays; +import java.util.List; import static com.mongodb.ReadPreference.primary; import static com.mongodb.ReadPreference.primaryPreferred; @@ -37,6 +38,7 @@ import static org.hamcrest.CoreMatchers.sameInstance; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assume.assumeFalse; @@ -461,6 +463,110 @@ public void shouldNotThrowAnExceptionOnCommandFailure() { assertThat(commandResult, hasFields(new String[]{"serverUsed", "ok", "errmsg"})); } + @Test + public void shouldAddReadOnlyUser() { + String userName = "newUser"; + String pwd = "pwd"; + getDatabase().addUser(userName, pwd.toCharArray(), true); + try { + assertCorrectUserExists(userName, pwd, true, getDatabase()); + } finally { + getDatabase().removeUser(userName); + } + } + + @Test + public void shouldAddReadWriteUser() { + String userName = "newUser"; + String pwd = "pwd"; + getDatabase().addUser(userName, pwd.toCharArray(), false); + try { + assertCorrectUserExists(userName, pwd, false, getDatabase()); + } finally { + getDatabase().removeUser(userName); + } + } + + @Test + public void shouldAddReadWriteAdminUser() throws UnknownHostException { + String userName = "newUser"; + String pwd = "pwd"; + MongoClient mongoClient = new MongoClient(getMongoClientURI()); + DB adminDB = mongoClient.getDB("admin"); + adminDB.addUser(userName, pwd.toCharArray(), false); + try { + assertTrue(adminDB.authenticate(userName, pwd.toCharArray())); + assertCorrectUserExists(userName, pwd, false, adminDB); + } finally { + adminDB.removeUser(userName); + mongoClient.close(); + } + } + + + @Test + public void shouldAddReadOnlyAdminUser() throws UnknownHostException { + String readWriteUserName = "newUserReadWrite"; + String readOnlyUserName = "newUser"; + String pwd = "pwd"; + MongoClient mongoClient = new MongoClient(getMongoClientURI()); + DB adminDB = mongoClient.getDB("admin"); + adminDB.addUser(readWriteUserName, pwd.toCharArray(), false); + adminDB.authenticate(readWriteUserName, pwd.toCharArray()); + adminDB.addUser(readOnlyUserName, pwd.toCharArray(), true); + try { + assertCorrectUserExists(readOnlyUserName, pwd, true, adminDB); + } finally { + adminDB.removeUser(readOnlyUserName); + adminDB.removeUser(readWriteUserName); + mongoClient.close(); + } + } + + @Test + public void shouldRemoveUser() { + String userName = "newUser"; + getDatabase().addUser(userName, "pwd".toCharArray(), true); + getDatabase().removeUser(userName); + assertThatUserIsRemoved(userName, getDatabase()); + } + + private void assertThatUserIsRemoved(final String userName, final DB database) { + if (serverIsAtLeastVersion(2.6)) { + CommandResult usersInfo = database.command(new BasicDBObject("usersInfo", userName)); + assertEquals(0, ((List) usersInfo.get("users")).size()); + } + else { + assertNull(database.getCollection("system.users").findOne(new BasicDBObject("user", userName))); + } + } + + + private void assertCorrectUserExists(final String userName, final String password, final boolean isReadOnly, final DB database) { + if (serverIsAtLeastVersion(2.6, database.getMongo())) { + CommandResult usersInfo = database.command(new BasicDBObject("usersInfo", userName)); + DBObject user = (DBObject) ((List) usersInfo.get("users")).get(0); + assertEquals(userName, user.get("user")); + assertEquals(database.getName(), user.get("db")); + assertEquals(getExpectedRole(isReadOnly, database), ((DBObject) ((List) user.get("roles")).get(0)).get("role")); + } + else { + assertEquals(new BasicDBObject("user", userName).append("readOnly", isReadOnly) + .append("pwd", getDatabase()._hash(userName, password.toCharArray())), + database.getCollection("system.users").findOne(new BasicDBObject("user", userName), + new BasicDBObject("_id", 0))); + } + } + + private String getExpectedRole(final boolean isReadOnly, final DB database) { + if (database.getName().equals("admin")) { + return isReadOnly ? "readAnyDatabase" : "root"; + } else { + return isReadOnly ? "read" : "dbOwner"; + } + } + + private DB getReplicaSetDB() throws UnknownHostException { Mongo mongo = new MongoClient(Arrays.asList(new ServerAddress("127.0.0.1"), new ServerAddress("127.0.0.1", 27018))); return mongo.getDB("database-" + System.nanoTime()); diff --git a/src/test/com/mongodb/Fixture.java b/src/test/com/mongodb/Fixture.java index 75b0b5f8058..b6231cc8b2a 100644 --- a/src/test/com/mongodb/Fixture.java +++ b/src/test/com/mongodb/Fixture.java @@ -17,6 +17,7 @@ package com.mongodb; import java.net.UnknownHostException; +import java.util.List; /** * Helper class for the acceptance tests. @@ -62,6 +63,39 @@ public static boolean serverIsAtLeastVersion(double version) { return Double.parseDouble(serverVersion.substring(0, 3)) >= version; } + public static boolean isStandalone() { + return !isReplicaSet() && !isSharded(); + } + + public static boolean isSharded() { + CommandResult isMasterResult = runIsMaster(); + Object msg = isMasterResult.get("msg"); + return msg != null && msg.equals("isdbgrid"); + } + + public static boolean isReplicaSet() { + return runIsMaster().get("setName") != null; + } + + public static boolean isServerStartedWithJournalingDisabled() { + return serverStartedWithBooleanOption("--nojournal", "nojournal"); + } + + private static boolean serverStartedWithBooleanOption(final String commandLineOption, final String configOption) { + CommandResult res = getMongoClient().getDB("admin").command(new BasicDBObject("getCmdLineOpts", 1)); + res.throwOnError(); + if (res.containsField("parsed") && ((DBObject) res.get("parsed")).containsField(configOption)) { + return (Boolean) ((DBObject) res.get("parsed")).get(configOption); + } else { + return ((List) res.get("argv")).contains(commandLineOption); + } + } + + private static CommandResult runIsMaster() { + // Check to see if this is a replica set... if not, get out of here. + return getMongoClient().getDB("admin").command(new BasicDBObject("ismaster", 1)); + } + static class ShutdownHook extends Thread { @Override public void run() { diff --git a/src/test/com/mongodb/JavaClientTest.java b/src/test/com/mongodb/JavaClientTest.java index d60a88f4f1f..7d5eacaa748 100644 --- a/src/test/com/mongodb/JavaClientTest.java +++ b/src/test/com/mongodb/JavaClientTest.java @@ -44,6 +44,7 @@ import java.util.concurrent.TimeUnit; import java.util.regex.Pattern; +import static java.util.Arrays.asList; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; @@ -491,7 +492,7 @@ public void testMapReduceInlineSecondary() throws Exception { return; } - Mongo mongo = new MongoClient(Arrays.asList(new ServerAddress("127.0.0.1", 27017), new ServerAddress("127.0.0.1", 27018)), + Mongo mongo = new MongoClient(asList(new ServerAddress("127.0.0.1", 27017), new ServerAddress("127.0.0.1", 27018)), MongoClientOptions.builder().writeConcern(WriteConcern.UNACKNOWLEDGED).build()); int size = getReplicaSetSize(mongo); @@ -914,6 +915,24 @@ public void testWriteResultOnUnacknowledgedUpdate(){ assertTrue(res.isLazy()); } + @Test + public void testWriteResultOnInsert(){ + WriteResult res = collection.insert(new BasicDBObject()); + assertEquals(0, res.getN()); + assertFalse(res.isUpdateOfExisting()); + assertNull(res.getUpsertedId()); + assertFalse(res.isLazy()); + } + + @Test + public void testWriteResultOnInsertList(){ + WriteResult res = collection.insert(Arrays.asList(new BasicDBObject(), new BasicDBObject())); + assertEquals(0, res.getN()); + assertFalse(res.isUpdateOfExisting()); + assertNull(res.getUpsertedId()); + assertFalse(res.isLazy()); + } + @Test public void testWriteResultOnUpdate(){ collection.insert(new BasicDBObject("_id", 1)); @@ -1101,7 +1120,7 @@ public void testBadKey(){ } catch (IllegalArgumentException e) {} try { - final List list = Arrays.asList(new BasicDBObject("$a", 1)); + final List list = asList(new BasicDBObject("$a", 1)); c.save(new BasicDBObject("a", list)); fail("Bad key was accepted"); } catch (IllegalArgumentException e) {} diff --git a/src/test/com/mongodb/MongoClientOptionsTest.java b/src/test/com/mongodb/MongoClientOptionsTest.java index 48ce26cb537..cc554cf2b2a 100644 --- a/src/test/com/mongodb/MongoClientOptionsTest.java +++ b/src/test/com/mongodb/MongoClientOptionsTest.java @@ -26,6 +26,7 @@ import static java.lang.System.getProperty; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; import static org.junit.Assert.fail; public class MongoClientOptionsTest { @@ -60,6 +61,7 @@ public void testBuilderDefaults() { assertEquals(getProperty("com.mongodb.slaveAcceptableLatencyMS") != null ? parseInt(getProperty("com.mongodb.slaveAcceptableLatencyMS")) : 15, options.getAcceptableLatencyDifference()); + assertNull(options.getRequiredReplicaSetName()); } @Test @@ -225,6 +227,7 @@ public void testBuilderBuild() { builder.heartbeatConnectTimeout(53); builder.heartbeatSocketTimeout(54); builder.heartbeatThreadCount(4); + builder.requiredReplicaSetName("test"); SocketFactory socketFactory = SSLSocketFactory.getDefault(); builder.socketFactory(socketFactory); @@ -269,6 +272,7 @@ public DBDecoder create() { assertEquals(socketFactory, options.getSocketFactory()); assertEquals(encoderFactory, options.getDbEncoderFactory()); assertEquals(decoderFactory, options.getDbDecoderFactory()); + assertEquals("test", options.getRequiredReplicaSetName()); } @Test diff --git a/src/test/com/mongodb/MongoClientURITest.java b/src/test/com/mongodb/MongoClientURITest.java index 7f23517e872..0cca48624b0 100644 --- a/src/test/com/mongodb/MongoClientURITest.java +++ b/src/test/com/mongodb/MongoClientURITest.java @@ -201,16 +201,22 @@ public void testSSLOption() { public void testOptions() { MongoClientURI uAmp = new MongoClientURI("mongodb://localhost/?" + "maxPoolSize=10&waitQueueMultiple=5&waitQueueTimeoutMS=150&" + + "minPoolSize=7&maxIdleTimeMS=1000&maxLifeTimeMS=2000&" + + "replicaSet=test&" + "connectTimeoutMS=2500&socketTimeoutMS=5500&autoConnectRetry=true&" + "slaveOk=true&safe=false&w=1&wtimeout=2500&fsync=true"); assertOnOptions(uAmp.getOptions()); MongoClientURI uSemi = new MongoClientURI("mongodb://localhost/?" + "maxPoolSize=10;waitQueueMultiple=5;waitQueueTimeoutMS=150;" + + "minPoolSize=7;maxIdleTimeMS=1000;maxLifeTimeMS=2000;" + + "replicaSet=test;" + "connectTimeoutMS=2500;socketTimeoutMS=5500;autoConnectRetry=true;" + "slaveOk=true;safe=false;w=1;wtimeout=2500;fsync=true"); assertOnOptions(uSemi.getOptions()); MongoClientURI uMixed = new MongoClientURI("mongodb://localhost/test?" + "maxPoolSize=10&waitQueueMultiple=5;waitQueueTimeoutMS=150;" + + "minPoolSize=7&maxIdleTimeMS=1000;maxLifeTimeMS=2000&" + + "replicaSet=test;" + "connectTimeoutMS=2500;socketTimeoutMS=5500&autoConnectRetry=true;" + "slaveOk=true;safe=false&w=1;wtimeout=2500;fsync=true"); assertOnOptions(uMixed.getOptions()); @@ -304,11 +310,15 @@ public void testMultipleIPV6ServersWithPorts() { @SuppressWarnings("deprecation") private void assertOnOptions(MongoClientOptions options) { assertEquals(10, options.getConnectionsPerHost(), 10); + assertEquals(7, options.getMinConnectionsPerHost()); + assertEquals(1000, options.getMaxConnectionIdleTime()); + assertEquals(2000, options.getMaxConnectionLifeTime()); assertEquals(5, options.getThreadsAllowedToBlockForConnectionMultiplier()); assertEquals(150, options.getMaxWaitTime()); assertEquals(5500, options.getSocketTimeout()); assertTrue(options.isAutoConnectRetry()); assertEquals(new WriteConcern(1, 2500, true), options.getWriteConcern()); assertEquals(ReadPreference.secondaryPreferred(), options.getReadPreference()); + assertEquals("test", options.getRequiredReplicaSetName()); } } diff --git a/src/test/com/mongodb/MongoOptionsTest.java b/src/test/com/mongodb/MongoOptionsTest.java index 9b1e05781fa..9bf0a978a98 100644 --- a/src/test/com/mongodb/MongoOptionsTest.java +++ b/src/test/com/mongodb/MongoOptionsTest.java @@ -61,6 +61,7 @@ public void testCopy() throws Exception { options.minConnectionsPerHost = 5; options.maxConnectionIdleTime = 50000; options.maxConnectionLifeTime = 500000; + options.requiredReplicaSetName = "set1"; final MongoOptions copy = options.copy(); assertEquals(options.connectionsPerHost, copy.connectionsPerHost); @@ -86,6 +87,7 @@ public void testCopy() throws Exception { assertEquals(options.minConnectionsPerHost, copy.minConnectionsPerHost); assertEquals(options.maxConnectionIdleTime, copy.maxConnectionIdleTime); assertEquals(options.maxConnectionLifeTime, copy.maxConnectionLifeTime); + assertEquals(options.requiredReplicaSetName, copy.requiredReplicaSetName); } @Test @@ -114,6 +116,7 @@ public void testGetterSetters() throws Exception { options.setReadPreference(ReadPreference.secondary()); options.setCursorFinalizerEnabled(true); options.setAlwaysUseMBeans(true); + options.requiredReplicaSetName = "set1"; assertEquals(options.getConnectionsPerHost(), 100); assertEquals(options.getThreadsAllowedToBlockForConnectionMultiplier(), 101); @@ -135,6 +138,7 @@ public void testGetterSetters() throws Exception { assertEquals(options.getReadPreference(), ReadPreference.secondary()); assertEquals(options.isCursorFinalizerEnabled(), true); assertEquals(options.isAlwaysUseMBeans(), true); + assertEquals(options.getRequiredReplicaSetName(), "set1"); } @Test diff --git a/src/test/com/mongodb/StickyHAShardedClusterServerSelectorTest.java b/src/test/com/mongodb/MongosHAServerSelectorTest.java similarity index 95% rename from src/test/com/mongodb/StickyHAShardedClusterServerSelectorTest.java rename to src/test/com/mongodb/MongosHAServerSelectorTest.java index cc12269c3f5..8736499ba4a 100644 --- a/src/test/com/mongodb/StickyHAShardedClusterServerSelectorTest.java +++ b/src/test/com/mongodb/MongosHAServerSelectorTest.java @@ -34,16 +34,16 @@ import static org.junit.Assert.assertEquals; -public class StickyHAShardedClusterServerSelectorTest { +public class MongosHAServerSelectorTest { - private StickyHAShardedClusterServerSelector selector; + private MongosHAServerSelector selector; private ServerDescription first; private ServerDescription secondConnecting; private ServerDescription secondConnected; @Before public void setUp() throws UnknownHostException { - selector = new StickyHAShardedClusterServerSelector(); + selector = new MongosHAServerSelector(); first = ServerDescription.builder() .state(Connected) .address(new ServerAddress()) diff --git a/src/test/com/mongodb/SecondaryReadTest.java b/src/test/com/mongodb/SecondaryReadTest.java index c2ff8ae1de1..ed19ea97c05 100644 --- a/src/test/com/mongodb/SecondaryReadTest.java +++ b/src/test/com/mongodb/SecondaryReadTest.java @@ -17,6 +17,7 @@ package com.mongodb; import com.mongodb.util.TestCase; +import org.junit.Ignore; import org.junit.Test; import java.util.ArrayList; @@ -36,6 +37,7 @@ public class SecondaryReadTest extends TestCase { * Assert that the percentage of reads to each secondary does not deviate by more than 1 % */ @Test + @Ignore public void testSecondaryReadBalance() throws Exception { if (!isReplicaSet(cleanupMongo)) { return; diff --git a/src/test/com/mongodb/WriteCommandHelperSpecification.groovy b/src/test/com/mongodb/WriteCommandHelperSpecification.groovy index 7ae38726531..c805a0ddcb0 100644 --- a/src/test/com/mongodb/WriteCommandHelperSpecification.groovy +++ b/src/test/com/mongodb/WriteCommandHelperSpecification.groovy @@ -22,6 +22,8 @@ import static com.mongodb.WriteCommandResultHelper.getBulkWriteException import static com.mongodb.WriteCommandResultHelper.getBulkWriteResult import static com.mongodb.WriteCommandResultHelper.hasError import static com.mongodb.WriteRequest.Type.INSERT +import static com.mongodb.WriteRequest.Type.REMOVE +import static com.mongodb.WriteRequest.Type.REPLACE import static com.mongodb.WriteRequest.Type.UPDATE class WriteCommandHelperSpecification extends Specification { @@ -41,6 +43,26 @@ class WriteCommandHelperSpecification extends Specification { .getUpserts() } + def 'should not have modified count for update with no nModified field in the result'() { + expect: + !getBulkWriteResult(UPDATE, getCommandResult(new BasicDBObject('n', 1))).isModifiedCountAvailable() + } + + def 'should not have modified count for replace with no nModified field in the result'() { + expect: + !getBulkWriteResult(REPLACE, getCommandResult(new BasicDBObject('n', 1))).isModifiedCountAvailable() + } + + def 'should have modified count of 0 for insert with no nModified field in the result'() { + expect: + 0 == getBulkWriteResult(INSERT, getCommandResult(new BasicDBObject('n', 1))).getModifiedCount() + } + + def 'should have modified count of 0 for remove with no nModified field in the result'() { + expect: + 0 == getBulkWriteResult(REMOVE, getCommandResult(new BasicDBObject('n', 1))).getModifiedCount() + } + def 'should not have error if writeErrors is empty and writeConcernError is missing'() { expect: !hasError(getCommandResult(new BasicDBObject())); diff --git a/src/test/com/mongodb/util/TestCase.java b/src/test/com/mongodb/util/TestCase.java index a1059694249..014a851da42 100644 --- a/src/test/com/mongodb/util/TestCase.java +++ b/src/test/com/mongodb/util/TestCase.java @@ -101,7 +101,11 @@ protected static DB getDatabase() { * @return true if server is at least specified version */ protected boolean serverIsAtLeastVersion(double version) { - String serverVersion = (String) cleanupMongo.getDB("admin").command("serverStatus").get("version"); + return serverIsAtLeastVersion(version, cleanupMongo); + } + + protected boolean serverIsAtLeastVersion(double version, Mongo mongo) { + String serverVersion = (String) mongo.getDB("admin").command("serverStatus").get("version"); return Double.parseDouble(serverVersion.substring(0, 3)) >= version; }