From f09c622b2f2035d981215c3592be05b26f0b5ba9 Mon Sep 17 00:00:00 2001 From: Christoph Strobl Date: Thu, 9 Jan 2020 13:13:41 +0100 Subject: [PATCH] DATAMONGO-2427 - Update GridFS API to use MongoDB native ByteBuffer Publisher API and introduce UUID format configuration options. Move from AsyncInputStream handling to Publisher for GridFS. UUID types require additional configuration setup to prevent errors while processing legacy (type 3) binary types. We still use type 3 as default but allow codec configuration for type 4 via Java and XML configuration. Updated migration guide. Original pull request: #823. --- README.adoc | 23 ++ .../data/mongodb/SpringDataMongoDB.java | 77 ++++ .../AbstractMongoClientConfiguration.java | 26 +- .../AbstractReactiveMongoConfiguration.java | 23 +- .../mongodb/config/MongoClientParser.java | 4 + .../config/MongoConfigurationSupport.java | 28 ++ .../mongodb/config/MongoParsingUtils.java | 19 + .../UUidRepresentationPropertyEditor.java | 45 +++ .../mongodb/core/MongoClientFactoryBean.java | 7 +- .../core/MongoClientSettingsFactoryBean.java | 15 +- .../gridfs/AsyncInputStreamAdapter.java | 361 ------------------ .../mongodb/gridfs/BinaryStreamAdapters.java | 80 ---- .../gridfs/DataBufferPublisherAdapter.java | 340 ----------------- .../gridfs/ReactiveGridFsOperations.java | 28 -- .../gridfs/ReactiveGridFsResource.java | 162 ++------ .../gridfs/ReactiveGridFsTemplate.java | 53 ++- .../query/StringBasedMongoQuery.java | 15 +- .../data/mongodb/config/spring-mongo-3.0.xsd | 20 + .../data/mongodb/SpringDataMongoDBTests.java | 36 ++ .../config/MongoClientNamespaceTests.java | 12 + .../config/MongoParserIntegrationTests.java | 1 - ...ngoClientSettingsFactoryBeanUnitTests.java | 17 + .../data/mongodb/core/MongoTemplateTests.java | 1 + .../gridfs/BinaryStreamAdaptersUnitTests.java | 103 ----- .../DataBufferPublisherAdapterUnitTests.java | 55 --- .../gridfs/ReactiveGridFsTemplateTests.java | 103 ++++- .../query/StringBasedMongoQueryUnitTests.java | 51 ++- .../mongodb/test/util/MongoTestUtils.java | 10 +- .../MongoClientNamespaceTests-context.xml | 4 + src/main/asciidoc/upgrading.adoc | 26 ++ 30 files changed, 581 insertions(+), 1164 deletions(-) create mode 100644 spring-data-mongodb/src/main/java/org/springframework/data/mongodb/SpringDataMongoDB.java create mode 100644 spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/UUidRepresentationPropertyEditor.java delete mode 100644 spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/AsyncInputStreamAdapter.java delete mode 100644 spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/BinaryStreamAdapters.java delete mode 100644 spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/DataBufferPublisherAdapter.java create mode 100644 spring-data-mongodb/src/test/java/org/springframework/data/mongodb/SpringDataMongoDBTests.java delete mode 100644 spring-data-mongodb/src/test/java/org/springframework/data/mongodb/gridfs/BinaryStreamAdaptersUnitTests.java delete mode 100644 spring-data-mongodb/src/test/java/org/springframework/data/mongodb/gridfs/DataBufferPublisherAdapterUnitTests.java diff --git a/README.adoc b/README.adoc index 820ace007..cbd524c70 100644 --- a/README.adoc +++ b/README.adoc @@ -208,6 +208,29 @@ Uses `MongoClientSettings` instead of `MongoClientOptions`. | |=== +=== Other Changes + +==== UUID Types + +The MongoDB UUID representation can now be configured with different formats. +This has to be done via `MongoClientSettings` as shown in the snippet below. + +.UUID Codec Configuration +==== +[source,java] +---- +static class Config extends AbstractMongoClientConfiguration { + + @Override + public void configureClientSettings(MongoClientSettings.Builder builder) { + builder.uuidRepresentation(UuidRepresentation.STANDARD); + } + + // ... +} +---- +==== + == Getting Help Having trouble with Spring Data? We’d love to help! diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/SpringDataMongoDB.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/SpringDataMongoDB.java new file mode 100644 index 000000000..cc3004644 --- /dev/null +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/SpringDataMongoDB.java @@ -0,0 +1,77 @@ +/* + * Copyright 2020 the original author or authors. + * + * 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 + * + * https://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 org.springframework.data.mongodb; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.data.util.Version; +import org.springframework.util.StringUtils; + +import com.mongodb.MongoDriverInformation; + +/** + * Class that exposes the SpringData MongoDB specific information like the current {@link Version} or + * {@link MongoDriverInformation driver information}. + * + * @author Christoph Strobl + * @since 3.0 + */ +public class SpringDataMongoDB { + + private static final Logger LOGGER = LoggerFactory.getLogger(SpringDataMongoDB.class); + + private static final Version FALLBACK_VERSION = new Version(3); + private static final MongoDriverInformation DRIVER_INFORMATION = MongoDriverInformation + .builder(MongoDriverInformation.builder().build()).driverName("spring-data").build(); + + /** + * Obtain the SpringData MongoDB specific driver information. + * + * @return never {@literal null}. + */ + public static MongoDriverInformation driverInformation() { + return DRIVER_INFORMATION; + } + + /** + * Fetches the "Implementation-Version" manifest attribute from the jar file. + *

+ * Note that some ClassLoaders do not expose the package metadata, hence this class might not be able to determine the + * version in all environments. In this case the current Major version is returned as a fallback. + * + * @return never {@literal null}. + */ + public static Version version() { + + Package pkg = SpringDataMongoDB.class.getPackage(); + String versionString = (pkg != null ? pkg.getImplementationVersion() : null); + + if (!StringUtils.hasText(versionString)) { + + LOGGER.debug("Unable to find Spring Data MongoDB version."); + return FALLBACK_VERSION; + } + + try { + return Version.parse(versionString); + } catch (Exception e) { + LOGGER.debug("Cannot read Spring Data MongoDB version '{}'.", versionString); + } + + return FALLBACK_VERSION; + } + +} diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/AbstractMongoClientConfiguration.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/AbstractMongoClientConfiguration.java index cce571498..2bac4e54a 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/AbstractMongoClientConfiguration.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/AbstractMongoClientConfiguration.java @@ -15,9 +15,11 @@ */ package org.springframework.data.mongodb.config; +import com.mongodb.MongoClientSettings.Builder; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.data.mongodb.MongoDatabaseFactory; +import org.springframework.data.mongodb.SpringDataMongoDB; import org.springframework.data.mongodb.core.MongoTemplate; import org.springframework.data.mongodb.core.SimpleMongoClientDatabaseFactory; import org.springframework.data.mongodb.core.convert.DbRefResolver; @@ -26,7 +28,9 @@ import org.springframework.data.mongodb.core.convert.MappingMongoConverter; import org.springframework.data.mongodb.core.mapping.Document; import org.springframework.lang.Nullable; +import com.mongodb.MongoClientSettings; import com.mongodb.client.MongoClient; +import com.mongodb.client.MongoClients; /** * Base class for Spring Data MongoDB configuration using JavaConfig with {@link com.mongodb.client.MongoClient}. @@ -40,11 +44,16 @@ public abstract class AbstractMongoClientConfiguration extends MongoConfiguratio /** * Return the {@link MongoClient} instance to connect to. Annotate with {@link Bean} in case you want to expose a - * {@link MongoClient} instance to the {@link org.springframework.context.ApplicationContext}. + * {@link MongoClient} instance to the {@link org.springframework.context.ApplicationContext}.
+ * Override {@link #mongoClientSettings()} to configure connection details. * - * @return + * @return never {@literal null}. + * @see #mongoClientSettings() + * @see #configureClientSettings(Builder) */ - public abstract MongoClient mongoClient(); + public MongoClient mongoClient() { + return createMongoClient(mongoClientSettings()); + } /** * Creates a {@link MongoTemplate}. @@ -107,4 +116,15 @@ public abstract class AbstractMongoClientConfiguration extends MongoConfiguratio return converter; } + + /** + * Create the Reactive Streams {@link com.mongodb.reactivestreams.client.MongoClient} instance with given + * {@link MongoClientSettings}. + * + * @return never {@literal null}. + * @since 3.0 + */ + protected MongoClient createMongoClient(MongoClientSettings settings) { + return MongoClients.create(settings, SpringDataMongoDB.driverInformation()); + } } diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/AbstractReactiveMongoConfiguration.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/AbstractReactiveMongoConfiguration.java index 2e550dfba..d64bcfc91 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/AbstractReactiveMongoConfiguration.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/AbstractReactiveMongoConfiguration.java @@ -18,13 +18,17 @@ package org.springframework.data.mongodb.config; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.data.mongodb.ReactiveMongoDatabaseFactory; +import org.springframework.data.mongodb.SpringDataMongoDB; import org.springframework.data.mongodb.core.ReactiveMongoOperations; import org.springframework.data.mongodb.core.ReactiveMongoTemplate; import org.springframework.data.mongodb.core.SimpleReactiveMongoDatabaseFactory; import org.springframework.data.mongodb.core.convert.MappingMongoConverter; import org.springframework.data.mongodb.core.convert.NoOpDbRefResolver; +import com.mongodb.MongoClientSettings; +import com.mongodb.MongoClientSettings.Builder; import com.mongodb.reactivestreams.client.MongoClient; +import com.mongodb.reactivestreams.client.MongoClients; /** * Base class for reactive Spring Data MongoDB configuration using JavaConfig. @@ -39,11 +43,16 @@ public abstract class AbstractReactiveMongoConfiguration extends MongoConfigurat /** * Return the Reactive Streams {@link MongoClient} instance to connect to. Annotate with {@link Bean} in case you want - * to expose a {@link MongoClient} instance to the {@link org.springframework.context.ApplicationContext}. + * to expose a {@link MongoClient} instance to the {@link org.springframework.context.ApplicationContext}.
+ * Override {@link #mongoClientSettings()} to configure connection details. * * @return never {@literal null}. + * @see #mongoClientSettings() + * @see #configureClientSettings(Builder) */ - public abstract MongoClient reactiveMongoClient(); + public MongoClient reactiveMongoClient() { + return createReactiveMongoClient(mongoClientSettings()); + } /** * Creates {@link ReactiveMongoOperations}. @@ -87,4 +96,14 @@ public abstract class AbstractReactiveMongoConfiguration extends MongoConfigurat return converter; } + + /** + * Create the Reactive Streams {@link MongoClient} instance with given {@link MongoClientSettings}. + * + * @return never {@literal null}. + * @since 3.0 + */ + protected MongoClient createReactiveMongoClient(MongoClientSettings settings) { + return MongoClients.create(settings, SpringDataMongoDB.driverInformation()); + } } diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/MongoClientParser.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/MongoClientParser.java index b6e13d856..b3dcfdabc 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/MongoClientParser.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/MongoClientParser.java @@ -87,6 +87,10 @@ public class MongoClientParser implements BeanDefinitionParser { .getComponent(MongoParsingUtils.getMongoCredentialPropertyEditor()); parserContext.registerBeanComponent(credentialsEditor); + BeanComponentDefinition uuidRepresentationEditor = helper + .getComponent(MongoParsingUtils.getUUidRepresentationEditorBuilder()); + parserContext.registerBeanComponent(uuidRepresentationEditor); + parserContext.popAndRegisterContainingComponent(); return mongoComponent.getBeanDefinition(); diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/MongoConfigurationSupport.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/MongoConfigurationSupport.java index ea137756f..a61646783 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/MongoConfigurationSupport.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/MongoConfigurationSupport.java @@ -20,6 +20,7 @@ import java.util.Collections; import java.util.HashSet; import java.util.Set; +import org.bson.UuidRepresentation; import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.ClassPathScanningCandidateComponentProvider; @@ -36,6 +37,9 @@ import org.springframework.data.mongodb.core.mapping.MongoMappingContext; import org.springframework.util.ClassUtils; import org.springframework.util.StringUtils; +import com.mongodb.MongoClientSettings; +import com.mongodb.MongoClientSettings.Builder; + /** * Base class for Spring Data MongoDB to be extended for JavaConfiguration usage. * @@ -184,4 +188,28 @@ public abstract class MongoConfigurationSupport { protected boolean autoIndexCreation() { return true; } + + /** + * Return the {@link MongoClientSettings} used to create the actual {@literal MongoClient}.
+ * Override either this method, or use {@link #configureClientSettings(Builder)} to alter the setup. + * + * @return never {@literal null}. + * @since 3.0 + */ + protected MongoClientSettings mongoClientSettings() { + + MongoClientSettings.Builder builder = MongoClientSettings.builder(); + configureClientSettings(builder); + return builder.build(); + } + + /** + * Configure {@link MongoClientSettings} via its {@link Builder} API. + * + * @param builder never {@literal null}. + * @since 3.0 + */ + protected void configureClientSettings(MongoClientSettings.Builder builder) { + builder.uuidRepresentation(UuidRepresentation.JAVA_LEGACY); + } } diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/MongoParsingUtils.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/MongoParsingUtils.java index 96d6a2320..7ed91b27f 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/MongoParsingUtils.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/MongoParsingUtils.java @@ -67,6 +67,7 @@ abstract class MongoParsingUtils { setPropertyValue(clientOptionsDefBuilder, settingsElement, "write-concern", "writeConcern"); setPropertyValue(clientOptionsDefBuilder, settingsElement, "retry-reads", "retryReads"); setPropertyValue(clientOptionsDefBuilder, settingsElement, "retry-writes", "retryWrites"); + setPropertyValue(clientOptionsDefBuilder, settingsElement, "uuid-representation", "uUidRepresentation"); // SocketSettings setPropertyValue(clientOptionsDefBuilder, settingsElement, "socket-connect-timeout", "socketConnectTimeoutMS"); @@ -223,4 +224,22 @@ abstract class MongoParsingUtils { return builder; } + /** + * Returns the {@link BeanDefinitionBuilder} to build a {@link BeanDefinition} for a + * {@link ConnectionStringPropertyEditor}. + * + * @return + * @since 3.0 + */ + static BeanDefinitionBuilder getUUidRepresentationEditorBuilder() { + + Map> customEditors = new ManagedMap<>(); + customEditors.put("org.bson.UuidRepresentation", UUidRepresentationPropertyEditor.class); + + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(CustomEditorConfigurer.class); + builder.addPropertyValue("customEditors", customEditors); + + return builder; + } + } diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/UUidRepresentationPropertyEditor.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/UUidRepresentationPropertyEditor.java new file mode 100644 index 000000000..2499ecc48 --- /dev/null +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/config/UUidRepresentationPropertyEditor.java @@ -0,0 +1,45 @@ +/* + * Copyright 2020 the original author or authors. + * + * 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 + * + * https://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 org.springframework.data.mongodb.config; + +import java.beans.PropertyEditorSupport; + +import org.bson.UuidRepresentation; +import org.springframework.lang.Nullable; +import org.springframework.util.StringUtils; + +/** + * Parse a {@link String} to a {@link UuidRepresentation}. + * + * @author Christoph Strobl + * @since 3.0 + */ +public class UUidRepresentationPropertyEditor extends PropertyEditorSupport { + + /* + * (non-Javadoc) + * @see java.beans.PropertyEditorSupport#setAsText(java.lang.String) + */ + @Override + public void setAsText(@Nullable String value) { + + if (!StringUtils.hasText(value)) { + return; + } + + setValue(UuidRepresentation.valueOf(value)); + } +} diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoClientFactoryBean.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoClientFactoryBean.java index d4ec1f5d8..7c281c7bf 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoClientFactoryBean.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoClientFactoryBean.java @@ -23,9 +23,11 @@ import java.util.function.Consumer; import java.util.function.Function; import java.util.stream.Collectors; +import org.bson.UuidRepresentation; import org.springframework.beans.factory.config.AbstractFactoryBean; import org.springframework.dao.DataAccessException; import org.springframework.dao.support.PersistenceExceptionTranslator; +import org.springframework.data.mongodb.SpringDataMongoDB; import org.springframework.lang.Nullable; import org.springframework.util.CollectionUtils; import org.springframework.util.ObjectUtils; @@ -164,6 +166,7 @@ public class MongoClientFactoryBean extends AbstractFactoryBean imp getOrDefault(port, "" + ServerAddress.defaultPort()))); Builder builder = MongoClientSettings.builder().applyConnectionString(connectionString); + builder.uuidRepresentation(UuidRepresentation.JAVA_LEGACY); if (mongoClientSettings != null) { @@ -291,6 +294,8 @@ public class MongoClientFactoryBean extends AbstractFactoryBean imp mongoClientSettings.getRetryReads(), connectionString.getRetryReads())); applySettings(builder::retryWrites, computeSettingsValue(defaultSettings.getRetryWrites(), mongoClientSettings.getRetryWrites(), connectionString.getRetryWritesValue())); + applySettings(builder::uuidRepresentation, + computeSettingsValue(null, mongoClientSettings.getUuidRepresentation(), UuidRepresentation.JAVA_LEGACY)); } if (!CollectionUtils.isEmpty(credential)) { @@ -344,7 +349,7 @@ public class MongoClientFactoryBean extends AbstractFactoryBean imp } private MongoClient createMongoClient(MongoClientSettings settings) throws UnknownHostException { - return MongoClients.create(settings); + return MongoClients.create(settings, SpringDataMongoDB.driverInformation()); } private String getOrDefault(Object value, String defaultValue) { diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoClientSettingsFactoryBean.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoClientSettingsFactoryBean.java index f8cf254cb..2c66fe381 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoClientSettingsFactoryBean.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoClientSettingsFactoryBean.java @@ -23,8 +23,8 @@ import java.util.concurrent.TimeUnit; import javax.net.ssl.SSLContext; +import org.bson.UuidRepresentation; import org.bson.codecs.configuration.CodecRegistry; - import org.springframework.beans.factory.config.AbstractFactoryBean; import org.springframework.lang.Nullable; import org.springframework.util.CollectionUtils; @@ -64,6 +64,8 @@ public class MongoClientSettingsFactoryBean extends AbstractFactoryBean Socket Settings private int socketConnectTimeoutMS = DEFAULT_MONGO_SETTINGS.getSocketSettings() @@ -378,6 +380,13 @@ public class MongoClientSettingsFactoryBean extends AbstractFactoryBean - * This adapter subscribes to the binary {@link Publisher} as soon as the first chunk gets {@link #read(ByteBuffer) - * requested}. Requests are queued and binary chunks are requested from the {@link Publisher}. As soon as the - * {@link Publisher} emits items, chunks are provided to the read request which completes the number-of-written-bytes - * {@link Publisher}. - *

- * {@link AsyncInputStream} is supposed to work as sequential callback API that is called until reaching EOF. - * {@link #close()} is propagated as cancellation signal to the binary {@link Publisher}. - * - * @author Mark Paluch - * @author Christoph Strobl - * @since 2.2 - */ -@RequiredArgsConstructor -class AsyncInputStreamAdapter implements AsyncInputStream { - - private static final AtomicLongFieldUpdater DEMAND = AtomicLongFieldUpdater - .newUpdater(AsyncInputStreamAdapter.class, "demand"); - - private static final AtomicIntegerFieldUpdater SUBSCRIBED = AtomicIntegerFieldUpdater - .newUpdater(AsyncInputStreamAdapter.class, "subscribed"); - - private static final int SUBSCRIPTION_NOT_SUBSCRIBED = 0; - private static final int SUBSCRIPTION_SUBSCRIBED = 1; - - private final Publisher buffers; - private final Context subscriberContext; - - private volatile Subscription subscription; - private volatile boolean cancelled; - private volatile boolean allDataBuffersReceived; - private volatile Throwable error; - private final Queue readRequests = Queues. small().get(); - - private final Queue bufferQueue = Queues. small().get(); - - // see DEMAND - volatile long demand; - - // see SUBSCRIBED - volatile int subscribed = SUBSCRIPTION_NOT_SUBSCRIBED; - - /* - * (non-Javadoc) - * @see com.mongodb.reactivestreams.client.gridfs.AsyncInputStream#read(java.nio.ByteBuffer) - */ - @Override - public Publisher read(ByteBuffer dst) { - - return Flux.create(sink -> { - - readRequests.offer(new ReadRequest(sink, dst)); - - sink.onCancel(this::terminatePendingReads); - sink.onDispose(this::terminatePendingReads); - sink.onRequest(this::request); - }); - } - - void onError(FluxSink sink, Throwable e) { - - readRequests.poll(); - sink.error(e); - } - - void onComplete(FluxSink sink, int writtenBytes) { - - readRequests.poll(); - DEMAND.decrementAndGet(this); - sink.next(writtenBytes); - sink.complete(); - } - - /* - * (non-Javadoc) - * @see com.mongodb.reactivestreams.client.gridfs.AsyncInputStream#skip(long) - */ - @Override - public Publisher skip(long bytesToSkip) { - throw new UnsupportedOperationException("Skip is currently not implemented"); - } - - /* - * (non-Javadoc) - * @see com.mongodb.reactivestreams.client.gridfs.AsyncInputStream#close() - */ - @Override - public Publisher close() { - - return Mono.create(sink -> { - - cancelled = true; - - if (error != null) { - terminatePendingReads(); - sink.error(error); - return; - } - - terminatePendingReads(); - sink.success(); - }); - } - - protected void request(long n) { - - if (allDataBuffersReceived && bufferQueue.isEmpty()) { - - terminatePendingReads(); - return; - } - - Operators.addCap(DEMAND, this, n); - - if (SUBSCRIBED.get(this) == SUBSCRIPTION_NOT_SUBSCRIBED) { - - if (SUBSCRIBED.compareAndSet(this, SUBSCRIPTION_NOT_SUBSCRIBED, SUBSCRIPTION_SUBSCRIBED)) { - buffers.subscribe(new DataBufferCoreSubscriber()); - } - - } else { - - Subscription subscription = this.subscription; - - if (subscription != null) { - requestFromSubscription(subscription); - } - } - - } - - void requestFromSubscription(Subscription subscription) { - - if (cancelled) { - subscription.cancel(); - } - - drainLoop(); - } - - void drainLoop() { - - while (DEMAND.get(AsyncInputStreamAdapter.this) > 0) { - - DataBuffer wip = bufferQueue.peek(); - - if (wip == null) { - break; - } - - if (wip.readableByteCount() == 0) { - bufferQueue.poll(); - continue; - } - - ReadRequest consumer = AsyncInputStreamAdapter.this.readRequests.peek(); - if (consumer == null) { - break; - } - - consumer.transferBytes(wip, wip.readableByteCount()); - } - - if (bufferQueue.isEmpty()) { - - if (allDataBuffersReceived) { - terminatePendingReads(); - return; - } - - if (demand > 0) { - subscription.request(1); - } - } - } - - /** - * Terminates pending reads with empty buffers. - */ - void terminatePendingReads() { - - ReadRequest readers; - - while ((readers = readRequests.poll()) != null) { - readers.onComplete(); - } - } - - private class DataBufferCoreSubscriber implements CoreSubscriber { - - @Override - public Context currentContext() { - return AsyncInputStreamAdapter.this.subscriberContext; - } - - @Override - public void onSubscribe(Subscription s) { - - AsyncInputStreamAdapter.this.subscription = s; - s.request(1); - } - - @Override - public void onNext(DataBuffer dataBuffer) { - - if (cancelled || allDataBuffersReceived) { - DataBufferUtils.release(dataBuffer); - Operators.onNextDropped(dataBuffer, AsyncInputStreamAdapter.this.subscriberContext); - return; - } - - ReadRequest readRequest = AsyncInputStreamAdapter.this.readRequests.peek(); - - if (readRequest == null) { - - DataBufferUtils.release(dataBuffer); - Operators.onNextDropped(dataBuffer, AsyncInputStreamAdapter.this.subscriberContext); - subscription.cancel(); - return; - } - - bufferQueue.offer(dataBuffer); - - drainLoop(); - } - - @Override - public void onError(Throwable t) { - - if (AsyncInputStreamAdapter.this.cancelled || AsyncInputStreamAdapter.this.allDataBuffersReceived) { - Operators.onErrorDropped(t, AsyncInputStreamAdapter.this.subscriberContext); - return; - } - - AsyncInputStreamAdapter.this.error = t; - AsyncInputStreamAdapter.this.allDataBuffersReceived = true; - terminatePendingReads(); - } - - @Override - public void onComplete() { - - AsyncInputStreamAdapter.this.allDataBuffersReceived = true; - if (bufferQueue.isEmpty()) { - terminatePendingReads(); - } - } - } - - /** - * Request to read bytes and transfer these to the associated {@link ByteBuffer}. - */ - class ReadRequest { - - private final FluxSink sink; - private final ByteBuffer dst; - - private int writtenBytes; - - ReadRequest(FluxSink sink, ByteBuffer dst) { - this.sink = sink; - this.dst = dst; - this.writtenBytes = -1; - } - - public void onComplete() { - - if (error != null) { - AsyncInputStreamAdapter.this.onError(sink, error); - return; - } - - AsyncInputStreamAdapter.this.onComplete(sink, writtenBytes); - } - - public void transferBytes(DataBuffer db, int bytes) { - - try { - - if (error != null) { - AsyncInputStreamAdapter.this.onError(sink, error); - return; - } - - ByteBuffer byteBuffer = db.asByteBuffer(); - int remaining = byteBuffer.remaining(); - int writeCapacity = Math.min(dst.remaining(), remaining); - int limit = Math.min(byteBuffer.position() + writeCapacity, byteBuffer.capacity()); - int toWrite = limit - byteBuffer.position(); - - if (toWrite == 0) { - - AsyncInputStreamAdapter.this.onComplete(sink, writtenBytes); - return; - } - - int oldPosition = byteBuffer.position(); - - byteBuffer.limit(toWrite); - dst.put(byteBuffer); - byteBuffer.limit(byteBuffer.capacity()); - byteBuffer.position(oldPosition); - db.readPosition(db.readPosition() + toWrite); - - if (writtenBytes == -1) { - writtenBytes = bytes; - } else { - writtenBytes += bytes; - } - - } catch (Exception e) { - AsyncInputStreamAdapter.this.onError(sink, e); - } finally { - - if (db.readableByteCount() == 0) { - DataBufferUtils.release(db); - } - } - } - } -} diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/BinaryStreamAdapters.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/BinaryStreamAdapters.java deleted file mode 100644 index e793f84c1..000000000 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/BinaryStreamAdapters.java +++ /dev/null @@ -1,80 +0,0 @@ -/* - * Copyright 2019-2020 the original author or authors. - * - * 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 - * - * https://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 org.springframework.data.mongodb.gridfs; - -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; - -import org.reactivestreams.Publisher; -import org.springframework.core.io.buffer.DataBuffer; -import org.springframework.core.io.buffer.DataBufferFactory; -import org.springframework.core.io.buffer.DataBufferUtils; - -import com.mongodb.reactivestreams.client.gridfs.AsyncInputStream; - -/** - * Utility methods to create adapters from between {@link Publisher} of {@link DataBuffer} and {@link AsyncInputStream}. - * - * @author Mark Paluch - * @since 2.2 - */ -class BinaryStreamAdapters { - - /** - * Creates a {@link Flux} emitting {@link DataBuffer} by reading binary chunks from {@link AsyncInputStream}. - * Publisher termination (completion, error, cancellation) closes the {@link AsyncInputStream}. - *

- * The resulting {@link org.reactivestreams.Publisher} filters empty binary chunks and uses {@link DataBufferFactory} - * settings to determine the chunk size. - * - * @param inputStream must not be {@literal null}. - * @param dataBufferFactory must not be {@literal null}. - * @param bufferSize read {@code n} bytes per iteration. - * @return {@link Flux} emitting {@link DataBuffer}s. - * @see DataBufferFactory#allocateBuffer() - */ - static Flux toPublisher(AsyncInputStream inputStream, DataBufferFactory dataBufferFactory, - int bufferSize) { - - return DataBufferPublisherAdapter.createBinaryStream(inputStream, dataBufferFactory, bufferSize) // - .filter(it -> { - - if (it.readableByteCount() == 0) { - DataBufferUtils.release(it); - return false; - } - return true; - }); - } - - /** - * Creates a {@link Mono} emitting a {@link AsyncInputStream} to consume a {@link Publisher} emitting - * {@link DataBuffer} and exposing the binary stream through {@link AsyncInputStream}. {@link DataBuffer}s are - * released by the adapter during consumption. - *

- * This method returns a {@link Mono} to retain the {@link reactor.util.context.Context subscriber context}. - * - * @param dataBuffers must not be {@literal null}. - * @return {@link Mono} emitting {@link AsyncInputStream}. - * @see DataBufferUtils#release(DataBuffer) - */ - static Mono toAsyncInputStream(Publisher dataBuffers) { - - return Mono.create(sink -> { - sink.success(new AsyncInputStreamAdapter(dataBuffers, sink.currentContext())); - }); - } -} diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/DataBufferPublisherAdapter.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/DataBufferPublisherAdapter.java deleted file mode 100644 index 359d0fb05..000000000 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/DataBufferPublisherAdapter.java +++ /dev/null @@ -1,340 +0,0 @@ -/* - * Copyright 2019-2020 the original author or authors. - * - * 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 - * - * https://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 org.springframework.data.mongodb.gridfs; - -import lombok.RequiredArgsConstructor; -import reactor.core.CoreSubscriber; -import reactor.core.publisher.Flux; -import reactor.core.publisher.FluxSink; -import reactor.core.publisher.Mono; -import reactor.core.publisher.Operators; -import reactor.util.context.Context; - -import java.nio.ByteBuffer; -import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; -import java.util.concurrent.atomic.AtomicLongFieldUpdater; - -import org.reactivestreams.Publisher; -import org.reactivestreams.Subscription; -import org.springframework.core.io.buffer.DataBuffer; -import org.springframework.core.io.buffer.DataBufferFactory; - -import com.mongodb.reactivestreams.client.gridfs.AsyncInputStream; - -/** - * Utility to adapt a {@link AsyncInputStream} to a {@link Publisher} emitting {@link DataBuffer}. - * - * @author Mark Paluch - * @author Christoph Strobl - * @since 2.2 - */ -class DataBufferPublisherAdapter { - - /** - * Creates a {@link Publisher} emitting {@link DataBuffer}s by reading binary chunks from {@link AsyncInputStream}. - * Closes the {@link AsyncInputStream} once the {@link Publisher} terminates. - * - * @param inputStream must not be {@literal null}. - * @param dataBufferFactory must not be {@literal null}. - * @param bufferSize read {@code n} bytes per iteration. - * @return the resulting {@link Publisher}. - */ - static Flux createBinaryStream(AsyncInputStream inputStream, DataBufferFactory dataBufferFactory, - int bufferSize) { - - return Flux.usingWhen(Mono.just(new DelegatingAsyncInputStream(inputStream, dataBufferFactory, bufferSize)), - DataBufferPublisherAdapter::doRead, AsyncInputStream::close, (it, err) -> it.close(), AsyncInputStream::close); - } - - /** - * Use an {@link AsyncInputStreamHandler} to read data from the given {@link AsyncInputStream}. - * - * @param inputStream the source stream. - * @return a {@link Flux} emitting data chunks one by one. - * @since 2.2.1 - */ - private static Flux doRead(DelegatingAsyncInputStream inputStream) { - - AsyncInputStreamHandler streamHandler = new AsyncInputStreamHandler(inputStream, inputStream.dataBufferFactory, - inputStream.bufferSize); - return Flux.create((sink) -> { - - sink.onDispose(streamHandler::close); - sink.onCancel(streamHandler::close); - - sink.onRequest(n -> { - streamHandler.request(sink, n); - }); - }); - } - - /** - * An {@link AsyncInputStream} also holding a {@link DataBufferFactory} and default {@literal bufferSize} for reading - * from it, delegating operations on the {@link AsyncInputStream} to the reference instance.
- * Used to pass on the {@link AsyncInputStream} and parameters to avoid capturing lambdas. - * - * @author Christoph Strobl - * @since 2.2.1 - */ - private static class DelegatingAsyncInputStream implements AsyncInputStream { - - private final AsyncInputStream inputStream; - private final DataBufferFactory dataBufferFactory; - private int bufferSize; - - /** - * @param inputStream the source input stream. - * @param dataBufferFactory - * @param bufferSize - */ - DelegatingAsyncInputStream(AsyncInputStream inputStream, DataBufferFactory dataBufferFactory, int bufferSize) { - - this.inputStream = inputStream; - this.dataBufferFactory = dataBufferFactory; - this.bufferSize = bufferSize; - } - - /* - * (non-Javadoc) - * @see com.mongodb.reactivestreams.client.gridfs.AsyncInputStream#read(java.nio.ByteBuffer) - */ - @Override - public Publisher read(ByteBuffer dst) { - return inputStream.read(dst); - } - - /* - * (non-Javadoc) - * @see com.mongodb.reactivestreams.client.gridfs.AsyncInputStream#skip(long) - */ - @Override - public Publisher skip(long bytesToSkip) { - return inputStream.skip(bytesToSkip); - } - - /* - * (non-Javadoc) - * @see com.mongodb.reactivestreams.client.gridfs.AsyncInputStream#close() - */ - @Override - public Publisher close() { - return inputStream.close(); - } - } - - @RequiredArgsConstructor - static class AsyncInputStreamHandler { - - private static final AtomicLongFieldUpdater DEMAND = AtomicLongFieldUpdater - .newUpdater(AsyncInputStreamHandler.class, "demand"); - - private static final AtomicIntegerFieldUpdater STATE = AtomicIntegerFieldUpdater - .newUpdater(AsyncInputStreamHandler.class, "state"); - - private static final AtomicIntegerFieldUpdater DRAIN = AtomicIntegerFieldUpdater - .newUpdater(AsyncInputStreamHandler.class, "drain"); - - private static final AtomicIntegerFieldUpdater READ = AtomicIntegerFieldUpdater - .newUpdater(AsyncInputStreamHandler.class, "read"); - - private static final int STATE_OPEN = 0; - private static final int STATE_CLOSED = 1; - - private static final int DRAIN_NONE = 0; - private static final int DRAIN_COMPLETION = 1; - - private static final int READ_NONE = 0; - private static final int READ_IN_PROGRESS = 1; - - final AsyncInputStream inputStream; - final DataBufferFactory dataBufferFactory; - final int bufferSize; - - // see DEMAND - volatile long demand; - - // see STATE - volatile int state = STATE_OPEN; - - // see DRAIN - volatile int drain = DRAIN_NONE; - - // see READ_IN_PROGRESS - volatile int read = READ_NONE; - - void request(FluxSink sink, long n) { - - Operators.addCap(DEMAND, this, n); - drainLoop(sink); - } - - /** - * Loops while we have demand and while no read is in progress. - * - * @param sink - */ - void drainLoop(FluxSink sink) { - while (onShouldRead()) { - emitNext(sink); - } - } - - boolean onShouldRead() { - return !isClosed() && getDemand() > 0 && onWantRead(); - } - - boolean onWantRead() { - return READ.compareAndSet(this, READ_NONE, READ_IN_PROGRESS); - } - - void onReadDone() { - READ.compareAndSet(this, READ_IN_PROGRESS, READ_NONE); - } - - long getDemand() { - return DEMAND.get(this); - } - - void decrementDemand() { - DEMAND.decrementAndGet(this); - } - - void close() { - STATE.compareAndSet(this, STATE_OPEN, STATE_CLOSED); - } - - boolean enterDrainLoop() { - return DRAIN.compareAndSet(this, DRAIN_NONE, DRAIN_COMPLETION); - } - - void leaveDrainLoop() { - DRAIN.set(this, DRAIN_NONE); - } - - boolean isClosed() { - return STATE.get(this) == STATE_CLOSED; - } - - /** - * Emit the next {@link DataBuffer}. - * - * @param sink - * @return - */ - private void emitNext(FluxSink sink) { - - ByteBuffer transport = ByteBuffer.allocate(bufferSize); - BufferCoreSubscriber bufferCoreSubscriber = new BufferCoreSubscriber(sink, dataBufferFactory, transport); - try { - inputStream.read(transport).subscribe(bufferCoreSubscriber); - } catch (Throwable e) { - sink.error(e); - } - } - - private class BufferCoreSubscriber implements CoreSubscriber { - - private final FluxSink sink; - private final DataBufferFactory factory; - private final ByteBuffer transport; - private volatile Subscription subscription; - - BufferCoreSubscriber(FluxSink sink, DataBufferFactory factory, ByteBuffer transport) { - - this.sink = sink; - this.factory = factory; - this.transport = transport; - } - - @Override - public Context currentContext() { - return sink.currentContext(); - } - - @Override - public void onSubscribe(Subscription s) { - - this.subscription = s; - s.request(1); - } - - @Override - public void onNext(Integer bytes) { - - if (isClosed()) { - return; - } - - if (bytes > 0) { - - DataBuffer buffer = readNextChunk(); - sink.next(buffer); - decrementDemand(); - } - - if (bytes == -1) { - sink.complete(); - return; - } - - subscription.request(1); - } - - private DataBuffer readNextChunk() { - - transport.flip(); - - DataBuffer dataBuffer = factory.allocateBuffer(transport.remaining()); - dataBuffer.write(transport); - - transport.clear(); - - return dataBuffer; - } - - @Override - public void onError(Throwable t) { - - if (isClosed()) { - - Operators.onErrorDropped(t, sink.currentContext()); - return; - } - - close(); - sink.error(t); - } - - @Override - public void onComplete() { - - onReadDone(); - - if (!isClosed()) { - - if (enterDrainLoop()) { - try { - drainLoop(sink); - } finally { - leaveDrainLoop(); - } - } - - } - } - } - } -} diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/ReactiveGridFsOperations.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/ReactiveGridFsOperations.java index 9170da7cc..f189ea9d8 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/ReactiveGridFsOperations.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/ReactiveGridFsOperations.java @@ -27,7 +27,6 @@ import org.springframework.data.mongodb.core.query.Query; import org.springframework.lang.Nullable; import com.mongodb.client.gridfs.model.GridFSFile; -import com.mongodb.reactivestreams.client.gridfs.AsyncInputStream; /** * Collection of operations to store and read files from MongoDB GridFS using reactive infrastructure. @@ -101,20 +100,6 @@ public interface ReactiveGridFsOperations { return store(content, filename, null, metadata); } - /** - * Stores the given content into a file with the given name and content type using the given metadata. The metadata - * object will be marshalled before writing. - * - * @param content must not be {@literal null}. - * @param filename must not be {@literal null} or empty. - * @param contentType can be {@literal null}. - * @param metadata can be {@literal null} - * @return a {@link Mono} emitting the {@link ObjectId} of the {@link com.mongodb.client.gridfs.model.GridFSFile} just - * created. - */ - Mono store(AsyncInputStream content, @Nullable String filename, @Nullable String contentType, - @Nullable Object metadata); - /** * Stores the given content into a file with the given name and content type using the given metadata. The metadata * object will be marshalled before writing. @@ -142,19 +127,6 @@ public interface ReactiveGridFsOperations { return store(content, filename, null, metadata); } - /** - * Stores the given content into a file with the given name and content type using the given metadata. - * - * @param content must not be {@literal null}. - * @param filename must not be {@literal null} or empty. - * @param contentType can be {@literal null}. - * @param metadata can be {@literal null}. - * @return a {@link Mono} emitting the {@link ObjectId} of the {@link com.mongodb.client.gridfs.model.GridFSFile} just - * created. - */ - Mono store(AsyncInputStream content, @Nullable String filename, @Nullable String contentType, - @Nullable Document metadata); - /** * Stores the given content into a file with the given name and content type using the given metadata. * diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/ReactiveGridFsResource.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/ReactiveGridFsResource.java index 173ab2400..14be071eb 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/ReactiveGridFsResource.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/ReactiveGridFsResource.java @@ -16,20 +16,17 @@ package org.springframework.data.mongodb.gridfs; import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; -import java.io.FileNotFoundException; -import java.io.IOException; -import java.io.InputStream; -import java.util.function.IntFunction; - -import org.reactivestreams.Publisher; -import org.springframework.core.io.AbstractResource; import org.springframework.core.io.Resource; import org.springframework.core.io.buffer.DataBuffer; +import org.springframework.core.io.buffer.DataBufferFactory; +import org.springframework.core.io.buffer.DefaultDataBufferFactory; import org.springframework.lang.Nullable; import org.springframework.util.Assert; import com.mongodb.client.gridfs.model.GridFSFile; +import com.mongodb.reactivestreams.client.gridfs.GridFSDownloadPublisher; /** * Reactive {@link GridFSFile} based {@link Resource} implementation. @@ -38,13 +35,10 @@ import com.mongodb.client.gridfs.model.GridFSFile; * @author Christoph Strobl * @since 2.2 */ -public class ReactiveGridFsResource extends AbstractResource { +public class ReactiveGridFsResource { - private static final Integer DEFAULT_CHUNK_SIZE = 256 * 1024; - - private final @Nullable GridFSFile file; + private final GridFSDownloadPublisher content; private final String filename; - private final IntFunction> contentFunction; /** * Creates a new, absent {@link ReactiveGridFsResource}. @@ -52,35 +46,10 @@ public class ReactiveGridFsResource extends AbstractResource { * @param filename filename of the absent resource. * @param content */ - private ReactiveGridFsResource(String filename, Publisher content) { + public ReactiveGridFsResource(String filename, @Nullable GridFSDownloadPublisher content) { - this.file = null; + this.content = content; this.filename = filename; - this.contentFunction = any -> Flux.from(content); - } - - /** - * Creates a new {@link ReactiveGridFsResource} from the given {@link GridFSFile}. - * - * @param file must not be {@literal null}. - * @param content - */ - public ReactiveGridFsResource(GridFSFile file, Publisher content) { - this(file, (IntFunction>) any -> Flux.from(content)); - } - - /** - * Creates a new {@link ReactiveGridFsResource} from the given {@link GridFSFile}. - * - * @param file must not be {@literal null}. - * @param contentFunction - * @since 2.2.1 - */ - ReactiveGridFsResource(GridFSFile file, IntFunction> contentFunction) { - - this.file = file; - this.filename = file.getFilename(); - this.contentFunction = contentFunction; } /** @@ -93,123 +62,54 @@ public class ReactiveGridFsResource extends AbstractResource { public static ReactiveGridFsResource absent(String filename) { Assert.notNull(filename, "Filename must not be null"); - - return new ReactiveGridFsResource(filename, Flux.empty()); - } - - /* - * (non-Javadoc) - * @see org.springframework.core.io.InputStreamResource#getInputStream() - */ - @Override - public InputStream getInputStream() throws IllegalStateException { - throw new UnsupportedOperationException(); - } - - /* - * (non-Javadoc) - * @see org.springframework.core.io.AbstractResource#contentLength() - */ - @Override - public long contentLength() throws IOException { - - verifyExists(); - return getGridFSFile().getLength(); - } - - /* - * (non-Javadoc) - * @see org.springframework.core.io.AbstractResource#getFilename() - */ - @Override - public String getFilename() throws IllegalStateException { - return this.filename; - } - - /* - * (non-Javadoc) - * @see org.springframework.core.io.AbstractResource#exists() - */ - @Override - public boolean exists() { - return this.file != null; - } - - /* - * (non-Javadoc) - * @see org.springframework.core.io.AbstractResource#lastModified() - */ - @Override - public long lastModified() throws IOException { - - verifyExists(); - return getGridFSFile().getUploadDate().getTime(); - } - - /* - * (non-Javadoc) - * @see org.springframework.core.io.AbstractResource#getDescription() - */ - @Override - public String getDescription() { - return String.format("GridFs resource [%s]", this.getFilename()); + return new ReactiveGridFsResource(filename, null); } /** - * Returns the {@link Resource}'s id. - * - * @return never {@literal null}. - * @throws IllegalStateException if the file does not {@link #exists()}. + * @see org.springframework.core.io.AbstractResource#getFilename() */ - public Object getId() { - - Assert.state(exists(), () -> String.format("%s does not exist.", getDescription())); - - return getGridFSFile().getId(); + public String getFilename() throws IllegalStateException { + return this.filename; } /** * @return the underlying {@link GridFSFile}. Can be {@literal null} if absent. * @since 2.2 */ - @Nullable - public GridFSFile getGridFSFile() { - return file; + public Mono getGridFSFile() { + return content != null ? Mono.from(content.getGridFSFile()) : Mono.empty(); } /** - * Retrieve the download stream using the default chunk size of 256 kB. - * - * @return a {@link Flux} emitting data chunks one by one. Please make sure to - * {@link org.springframework.core.io.buffer.DataBufferUtils#release(DataBuffer) release} all - * {@link DataBuffer buffers} when done. */ public Flux getDownloadStream() { - return getDownloadStream(DEFAULT_CHUNK_SIZE); + + if (content == null) { + return Flux.empty(); + } + + return createDownloadStream(content); } /** - * Retrieve the download stream. - * - * @param chunkSize chunk size in bytes to use. - * @return a {@link Flux} emitting data chunks one by one. Please make sure to - * {@link org.springframework.core.io.buffer.DataBufferUtils#release(DataBuffer) release} all - * {@link DataBuffer buffers} when done. - * @since 2.2.1 */ public Flux getDownloadStream(int chunkSize) { - if (!exists()) { - return Flux.error(new FileNotFoundException(String.format("%s does not exist.", getDescription()))); + if (content == null) { + return Flux.empty(); + } - return contentFunction.apply(chunkSize); + return createDownloadStream(content.bufferSizeBytes(chunkSize)); } - private void verifyExists() throws FileNotFoundException { + private Flux createDownloadStream(GridFSDownloadPublisher publisher) { - if (!exists()) { - throw new FileNotFoundException(String.format("%s does not exist.", getDescription())); - } + DataBufferFactory bufferFactory = new DefaultDataBufferFactory(); + return Flux.from(publisher).map(bufferFactory::wrap); + } + + public boolean exists() { + return content != null; } } diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/ReactiveGridFsTemplate.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/ReactiveGridFsTemplate.java index 3537b146e..c7f3c8f31 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/ReactiveGridFsTemplate.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/ReactiveGridFsTemplate.java @@ -21,6 +21,8 @@ import static org.springframework.data.mongodb.gridfs.GridFsCriteria.*; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import java.nio.ByteBuffer; + import org.bson.Document; import org.bson.types.ObjectId; import org.reactivestreams.Publisher; @@ -37,12 +39,12 @@ import org.springframework.util.Assert; import org.springframework.util.StringUtils; import com.mongodb.client.gridfs.model.GridFSFile; +import com.mongodb.client.gridfs.model.GridFSUploadOptions; import com.mongodb.reactivestreams.client.MongoDatabase; -import com.mongodb.reactivestreams.client.gridfs.AsyncInputStream; import com.mongodb.reactivestreams.client.gridfs.GridFSBucket; import com.mongodb.reactivestreams.client.gridfs.GridFSBuckets; -import com.mongodb.reactivestreams.client.gridfs.GridFSDownloadStream; import com.mongodb.reactivestreams.client.gridfs.GridFSFindPublisher; +import com.mongodb.reactivestreams.client.gridfs.GridFSUploadPublisher; /** * {@link ReactiveGridFsOperations} implementation to store content into MongoDB GridFS. Uses by default @@ -51,6 +53,7 @@ import com.mongodb.reactivestreams.client.gridfs.GridFSFindPublisher; * @author Mark Paluch * @author Nick Stolwijk * @author Denis Zavedeev + * @author Christoph Strobl * @since 2.2 */ public class ReactiveGridFsTemplate extends GridFsOperationsSupport implements ReactiveGridFsOperations { @@ -105,16 +108,6 @@ public class ReactiveGridFsTemplate extends GridFsOperationsSupport implements R this.bucket = bucket; } - /* - * (non-Javadoc) - * @see org.springframework.data.mongodb.gridfs.ReactiveGridFsOperations#store(com.mongodb.reactivestreams.client.gridfs.AsyncInputStream, java.lang.String, java.lang.String, java.lang.Object) - */ - @Override - public Mono store(AsyncInputStream content, @Nullable String filename, @Nullable String contentType, - @Nullable Object metadata) { - return store(content, filename, contentType, toDocument(metadata)); - } - /* * (non-Javadoc) * @see org.springframework.data.mongodb.gridfs.ReactiveGridFsOperations#store(org.reactivestreams.Publisher, java.lang.String, java.lang.String, java.lang.Object) @@ -125,18 +118,6 @@ public class ReactiveGridFsTemplate extends GridFsOperationsSupport implements R return store(content, filename, contentType, toDocument(metadata)); } - /* - * (non-Javadoc) - * @see org.springframework.data.mongodb.gridfs.ReactiveGridFsOperations#store(com.mongodb.reactivestreams.client.gridfs.AsyncInputStream, java.lang.String, java.lang.String, org.bson.Document) - */ - @Override - public Mono store(AsyncInputStream content, @Nullable String filename, @Nullable String contentType, - @Nullable Document metadata) { - - Assert.notNull(content, "InputStream must not be null!"); - return Mono.from(getGridFs().uploadFromStream(filename, content, computeUploadOptionsFor(contentType, metadata))); - } - /* * (non-Javadoc) * @see org.springframework.data.mongodb.gridfs.ReactiveGridFsOperations#store(org.reactivestreams.Publisher, java.lang.String, java.lang.String, org.bson.Document) @@ -147,7 +128,12 @@ public class ReactiveGridFsTemplate extends GridFsOperationsSupport implements R Assert.notNull(content, "Content must not be null!"); - return BinaryStreamAdapters.toAsyncInputStream(content).flatMap(it -> store(it, filename, contentType, metadata)); + GridFSUploadOptions uploadOptions = new GridFSUploadOptions(); + uploadOptions.metadata(metadata); + + GridFSUploadPublisher publisher = getGridFs().uploadFromPublisher(filename, + Flux.from(content).map(this::dataBufferToByteBuffer), uploadOptions); + return Mono.from(publisher); } /* @@ -223,12 +209,7 @@ public class ReactiveGridFsTemplate extends GridFsOperationsSupport implements R Assert.notNull(file, "GridFSFile must not be null!"); return Mono.fromSupplier(() -> { - - return new ReactiveGridFsResource(file, chunkSize -> { - - GridFSDownloadStream stream = getGridFs().openDownloadStream(file.getId()); - return BinaryStreamAdapters.toPublisher(stream, dataBufferFactory, chunkSize); - }); + return new ReactiveGridFsResource(file.getFilename(), getGridFs().downloadToPublisher(file.getId())); }); } @@ -284,4 +265,14 @@ public class ReactiveGridFsTemplate extends GridFsOperationsSupport implements R MongoDatabase db = dbFactory.getMongoDatabase(); return bucket == null ? GridFSBuckets.create(db) : GridFSBuckets.create(db, bucket); } + + private ByteBuffer dataBufferToByteBuffer(DataBuffer buffer) { + + ByteBuffer byteBuffer = buffer.asByteBuffer(); + ByteBuffer copy = ByteBuffer.allocate(byteBuffer.remaining()); + byteBuffer.put(copy); + copy.flip(); + + return copy; + } } diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/query/StringBasedMongoQuery.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/query/StringBasedMongoQuery.java index ff84452d8..1eb4f66c3 100644 --- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/query/StringBasedMongoQuery.java +++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/query/StringBasedMongoQuery.java @@ -16,9 +16,9 @@ package org.springframework.data.mongodb.repository.query; import org.bson.Document; +import org.bson.codecs.configuration.CodecRegistry; import org.slf4j.Logger; import org.slf4j.LoggerFactory; - import org.springframework.data.mongodb.core.MongoOperations; import org.springframework.data.mongodb.core.query.BasicQuery; import org.springframework.data.mongodb.core.query.Query; @@ -28,6 +28,9 @@ import org.springframework.data.repository.query.QueryMethodEvaluationContextPro import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.util.Assert; +import com.mongodb.MongoClientSettings; +import com.mongodb.client.MongoDatabase; + /** * Query to use a plain JSON String to create the {@link Query} to actually execute. * @@ -40,11 +43,11 @@ public class StringBasedMongoQuery extends AbstractMongoQuery { private static final String COUNT_EXISTS_AND_DELETE = "Manually defined query for %s cannot be a count and exists or delete query at the same time!"; private static final Logger LOG = LoggerFactory.getLogger(StringBasedMongoQuery.class); - private static final ParameterBindingDocumentCodec CODEC = new ParameterBindingDocumentCodec(); private final String query; private final String fieldSpec; + private final ParameterBindingDocumentCodec codec; private final SpelExpressionParser expressionParser; private final QueryMethodEvaluationContextProvider evaluationContextProvider; @@ -106,6 +109,10 @@ public class StringBasedMongoQuery extends AbstractMongoQuery { this.isExistsQuery = false; this.isDeleteQuery = false; } + + CodecRegistry codecRegistry = mongoOperations.execute(MongoDatabase::getCodecRegistry); + this.codec = new ParameterBindingDocumentCodec( + codecRegistry != null ? codecRegistry : MongoClientSettings.getDefaultCodecRegistry()); } /* @@ -118,8 +125,8 @@ public class StringBasedMongoQuery extends AbstractMongoQuery { ParameterBindingContext bindingContext = new ParameterBindingContext((accessor::getBindableValue), expressionParser, () -> evaluationContextProvider.getEvaluationContext(getQueryMethod().getParameters(), accessor.getValues())); - Document queryObject = CODEC.decode(this.query, bindingContext); - Document fieldsObject = CODEC.decode(this.fieldSpec, bindingContext); + Document queryObject = codec.decode(this.query, bindingContext); + Document fieldsObject = codec.decode(this.fieldSpec, bindingContext); Query query = new BasicQuery(queryObject, fieldsObject).with(accessor.getSort()); diff --git a/spring-data-mongodb/src/main/resources/org/springframework/data/mongodb/config/spring-mongo-3.0.xsd b/spring-data-mongodb/src/main/resources/org/springframework/data/mongodb/config/spring-mongo-3.0.xsd index a6f5ceffd..e670b66cd 100644 --- a/spring-data-mongodb/src/main/resources/org/springframework/data/mongodb/config/spring-mongo-3.0.xsd +++ b/spring-data-mongodb/src/main/resources/org/springframework/data/mongodb/config/spring-mongo-3.0.xsd @@ -370,6 +370,16 @@ Reference to FactoryBean for com.mongodb.AutoEncryptionSettings - @since 2.2 + + + + + + + + + + @@ -471,6 +481,16 @@ The application name to use when connecting to MongoDB. Mainly used to identify ]]> + + + + + + + + values = definition.getPropertyValues().getPropertyValueList(); - values.forEach(System.out::println); assertThat(values.get(2).getValue()).isInstanceOf(BeanDefinition.class); BeanDefinition x = (BeanDefinition) values.get(2).getValue(); diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/MongoClientSettingsFactoryBeanUnitTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/MongoClientSettingsFactoryBeanUnitTests.java index ee93abf13..98732af8d 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/MongoClientSettingsFactoryBeanUnitTests.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/MongoClientSettingsFactoryBeanUnitTests.java @@ -17,11 +17,13 @@ package org.springframework.data.mongodb.core; import static org.assertj.core.api.Assertions.*; +import org.bson.UuidRepresentation; import org.junit.Test; import org.springframework.beans.factory.support.DefaultListableBeanFactory; import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.data.mongodb.config.ReadConcernPropertyEditor; import org.springframework.data.mongodb.config.ReadPreferencePropertyEditor; +import org.springframework.data.mongodb.config.UUidRepresentationPropertyEditor; import org.springframework.test.util.ReflectionTestUtils; import com.mongodb.ReadConcern; @@ -63,4 +65,19 @@ public class MongoClientSettingsFactoryBeanUnitTests { MongoClientSettingsFactoryBean bean = factory.getBean("&factory", MongoClientSettingsFactoryBean.class); assertThat(ReflectionTestUtils.getField(bean, "readConcern")).isEqualTo(ReadConcern.MAJORITY); } + + @Test // DATAMONGO-2427 + public void convertsUuidRepresentationCorrectly() { + + RootBeanDefinition definition = new RootBeanDefinition(MongoClientSettingsFactoryBean.class); + definition.getPropertyValues().addPropertyValue("uUidRepresentation", "STANDARD"); + + DefaultListableBeanFactory factory = new DefaultListableBeanFactory(); + factory.registerCustomEditor(ReadPreference.class, UUidRepresentationPropertyEditor.class); + + factory.registerBeanDefinition("factory", definition); + + MongoClientSettingsFactoryBean bean = factory.getBean("&factory", MongoClientSettingsFactoryBean.class); + assertThat(ReflectionTestUtils.getField(bean, "uUidRepresentation")).isEqualTo(UuidRepresentation.STANDARD); + } } diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/MongoTemplateTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/MongoTemplateTests.java index 4ec1a1bdc..217a86a42 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/MongoTemplateTests.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/MongoTemplateTests.java @@ -20,6 +20,7 @@ import static org.springframework.data.mongodb.core.query.Criteria.*; import static org.springframework.data.mongodb.core.query.Query.*; import static org.springframework.data.mongodb.core.query.Update.*; +import com.mongodb.MongoClientSettings; import lombok.AllArgsConstructor; import lombok.Data; import lombok.EqualsAndHashCode; diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/gridfs/BinaryStreamAdaptersUnitTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/gridfs/BinaryStreamAdaptersUnitTests.java deleted file mode 100644 index 86609ca8c..000000000 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/gridfs/BinaryStreamAdaptersUnitTests.java +++ /dev/null @@ -1,103 +0,0 @@ -/* - * Copyright 2019-2020 the original author or authors. - * - * 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 - * - * https://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 org.springframework.data.mongodb.gridfs; - -import static org.assertj.core.api.Assertions.*; - -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; -import reactor.test.StepVerifier; - -import java.io.IOException; -import java.nio.ByteBuffer; - -import org.junit.Test; - -import org.springframework.core.io.ClassPathResource; -import org.springframework.core.io.buffer.DataBuffer; -import org.springframework.core.io.buffer.DataBufferUtils; -import org.springframework.core.io.buffer.DefaultDataBufferFactory; -import org.springframework.util.StreamUtils; - -import com.mongodb.reactivestreams.client.gridfs.AsyncInputStream; -import com.mongodb.reactivestreams.client.gridfs.helpers.AsyncStreamHelper; - -/** - * Unit tests for {@link BinaryStreamAdapters}. - * - * @author Mark Paluch - */ -public class BinaryStreamAdaptersUnitTests { - - @Test // DATAMONGO-1855 - public void shouldAdaptAsyncInputStreamToDataBufferPublisher() throws IOException { - - ClassPathResource resource = new ClassPathResource("gridfs/gridfs.xml"); - - byte[] content = StreamUtils.copyToByteArray(resource.getInputStream()); - AsyncInputStream inputStream = AsyncStreamHelper.toAsyncInputStream(resource.getInputStream()); - - Flux dataBuffers = BinaryStreamAdapters.toPublisher(inputStream, new DefaultDataBufferFactory(), 256); - - DataBufferUtils.join(dataBuffers) // - .as(StepVerifier::create) // - .consumeNextWith(actual -> { - - byte[] actualContent = new byte[actual.readableByteCount()]; - actual.read(actualContent); - assertThat(actualContent).isEqualTo(content); - }) // - .verifyComplete(); - } - - @Test // DATAMONGO-1855 - public void shouldAdaptBinaryPublisherToAsyncInputStream() throws IOException { - - ClassPathResource resource = new ClassPathResource("gridfs/gridfs.xml"); - - byte[] content = StreamUtils.copyToByteArray(resource.getInputStream()); - - Flux dataBuffers = DataBufferUtils.readInputStream(resource::getInputStream, - new DefaultDataBufferFactory(), 10); - - AsyncInputStream inputStream = BinaryStreamAdapters.toAsyncInputStream(dataBuffers).block(); - ByteBuffer complete = readBuffer(inputStream); - - assertThat(complete).isEqualTo(ByteBuffer.wrap(content)); - } - - static ByteBuffer readBuffer(AsyncInputStream inputStream) { - - ByteBuffer complete = ByteBuffer.allocate(1024); - - boolean hasData = true; - while (hasData) { - - ByteBuffer chunk = ByteBuffer.allocate(100); - - Integer bytesRead = Mono.from(inputStream.read(chunk)).block(); - - chunk.flip(); - complete.put(chunk); - - hasData = bytesRead > -1; - } - - complete.flip(); - - return complete; - } -} diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/gridfs/DataBufferPublisherAdapterUnitTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/gridfs/DataBufferPublisherAdapterUnitTests.java deleted file mode 100644 index e1336364f..000000000 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/gridfs/DataBufferPublisherAdapterUnitTests.java +++ /dev/null @@ -1,55 +0,0 @@ -/* - * Copyright 2019-2020 the original author or authors. - * - * 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 - * - * https://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 org.springframework.data.mongodb.gridfs; - -import static org.mockito.Mockito.*; - -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; -import reactor.test.StepVerifier; - -import org.junit.Test; - -import org.springframework.core.io.buffer.DataBuffer; -import org.springframework.core.io.buffer.DataBufferFactory; -import org.springframework.core.io.buffer.DefaultDataBufferFactory; - -/** - * Unit tests for {@link DataBufferPublisherAdapter}. - * - * @author Mark Paluch - */ -public class DataBufferPublisherAdapterUnitTests { - - DataBufferFactory factory = new DefaultDataBufferFactory(); - - @Test // DATAMONGO-2230 - public void adapterShouldPropagateErrors() { - - AsyncInputStreamAdapter asyncInput = mock(AsyncInputStreamAdapter.class); - - when(asyncInput.read(any())).thenReturn(Mono.just(1), Mono.error(new IllegalStateException())); - when(asyncInput.close()).thenReturn(Mono.empty()); - - Flux binaryStream = DataBufferPublisherAdapter.createBinaryStream(asyncInput, factory, 256); - - StepVerifier.create(binaryStream, 0) // - .thenRequest(1) // - .expectNextCount(1) // - .thenRequest(1) // - .verifyError(IllegalStateException.class); - } -} diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/gridfs/ReactiveGridFsTemplateTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/gridfs/ReactiveGridFsTemplateTests.java index a444b99b5..b3d0e3983 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/gridfs/ReactiveGridFsTemplateTests.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/gridfs/ReactiveGridFsTemplateTests.java @@ -26,13 +26,16 @@ import reactor.test.StepVerifier; import java.io.IOException; import java.nio.ByteBuffer; +import java.util.Arrays; import org.bson.BsonObjectId; import org.bson.Document; import org.bson.types.ObjectId; import org.junit.Before; +import org.junit.Ignore; import org.junit.Test; import org.junit.runner.RunWith; +import org.reactivestreams.Publisher; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.core.io.ClassPathResource; import org.springframework.core.io.Resource; @@ -54,8 +57,11 @@ import org.springframework.util.StreamUtils; import com.mongodb.client.gridfs.model.GridFSFile; import com.mongodb.internal.HexUtils; -import com.mongodb.reactivestreams.client.gridfs.AsyncInputStream; -import com.mongodb.reactivestreams.client.gridfs.helpers.AsyncStreamHelper; +import com.mongodb.internal.connection.tlschannel.impl.ByteBufferUtil; +import com.mongodb.reactivestreams.client.gridfs.GridFSBucket; +import com.mongodb.reactivestreams.client.gridfs.GridFSBuckets; +import com.mongodb.reactivestreams.client.gridfs.GridFSUploadPublisher; +import com.mongodb.reactivestreams.client.internal.Publishers; /** * Integration tests for {@link ReactiveGridFsTemplate}. @@ -85,6 +91,7 @@ public class ReactiveGridFsTemplateTests { } @Test // DATAMONGO-1855 + @Ignore("https://jira.mongodb.org/browse/JAVARS-224") public void storesAndFindsSimpleDocument() { DefaultDataBufferFactory factory = new DefaultDataBufferFactory(); @@ -102,6 +109,7 @@ public class ReactiveGridFsTemplateTests { } @Test // DATAMONGO-1855 + @Ignore("https://jira.mongodb.org/browse/JAVARS-224") public void storesAndLoadsLargeFileCorrectly() { ByteBuffer buffer = ByteBuffer.allocate(1000 * 1000); // 1 mb @@ -160,13 +168,15 @@ public class ReactiveGridFsTemplateTests { // } @Test // DATAMONGO-1855 + @Ignore("https://jira.mongodb.org/browse/JAVARS-224") public void writesMetadataCorrectly() throws IOException { Document metadata = new Document("key", "value"); - AsyncInputStream stream = AsyncStreamHelper.toAsyncInputStream(resource.getInputStream()); + Flux source = DataBufferUtils.read(resource, new DefaultDataBufferFactory(), 256); + // AsyncInputStream stream = AsyncStreamHelper.toAsyncInputStream(resource.getInputStream()); - ObjectId reference = operations.store(stream, "foo.xml", "binary/octet-stream", metadata).block(); + ObjectId reference = operations.store(source, "foo.xml", "binary/octet-stream", metadata).block(); operations.find(query(whereMetaData("key").is("value"))) // .as(StepVerifier::create) // @@ -177,14 +187,14 @@ public class ReactiveGridFsTemplateTests { } @Test // DATAMONGO-1855 + @Ignore("https://jira.mongodb.org/browse/JAVARS-224") public void marshalsComplexMetadata() throws IOException { Metadata metadata = new Metadata(); metadata.version = "1.0"; - AsyncInputStream stream = AsyncStreamHelper.toAsyncInputStream(resource.getInputStream()); - - ObjectId reference = operations.store(stream, "foo.xml", "binary/octet-stream", metadata).block(); + Flux source = DataBufferUtils.read(resource, new DefaultDataBufferFactory(), 256); + ObjectId reference = operations.store(source, "foo.xml", "binary/octet-stream", metadata).block(); operations.find(query(whereMetaData("version").is("1.0"))) // .as(StepVerifier::create) // @@ -196,12 +206,12 @@ public class ReactiveGridFsTemplateTests { } @Test // DATAMONGO-1855 + @Ignore("https://jira.mongodb.org/browse/JAVARS-224") public void getResourceShouldRetrieveContentByIdentity() throws IOException { byte[] content = StreamUtils.copyToByteArray(resource.getInputStream()); - AsyncInputStream upload = AsyncStreamHelper.toAsyncInputStream(resource.getInputStream()); - - ObjectId reference = operations.store(upload, "foo.xml", null, null).block(); + Flux source = DataBufferUtils.read(resource, new DefaultDataBufferFactory(), 256); + ObjectId reference = operations.store(source, "foo.xml", null, null).block(); operations.findOne(query(where("_id").is(reference))).flatMap(operations::getResource) .flatMapMany(ReactiveGridFsResource::getDownloadStream) // @@ -218,11 +228,12 @@ public class ReactiveGridFsTemplateTests { } @Test // DATAMONGO-1855, DATAMONGO-2240 + @Ignore("https://jira.mongodb.org/browse/JAVARS-224") public void shouldEmitFirstEntryWhenFindFirstRetrievesMoreThanOneResult() throws IOException { - AsyncInputStream upload1 = AsyncStreamHelper.toAsyncInputStream(resource.getInputStream()); - AsyncInputStream upload2 = AsyncStreamHelper - .toAsyncInputStream(new ClassPathResource("gridfs/another-resource.xml").getInputStream()); + Flux upload1 = DataBufferUtils.read(resource, new DefaultDataBufferFactory(), 256); + Flux upload2 = DataBufferUtils.read(new ClassPathResource("gridfs/another-resource.xml"), + new DefaultDataBufferFactory(), 256); operations.store(upload1, "foo.xml", null, null).block(); operations.store(upload2, "foo2.xml", null, null).block(); @@ -244,16 +255,17 @@ public class ReactiveGridFsTemplateTests { .assertNext(actual -> { assertThat(actual.exists()).isFalse(); - assertThat(actual.getGridFSFile()).isNull(); + assertThat(actual.getGridFSFile()).isEqualTo(Mono.empty()); }).verifyComplete(); } @Test // DATAMONGO-1855 + @Ignore("https://jira.mongodb.org/browse/JAVARS-224") public void shouldEmitErrorWhenFindOneRetrievesMoreThanOneResult() throws IOException { - AsyncInputStream upload1 = AsyncStreamHelper.toAsyncInputStream(resource.getInputStream()); - AsyncInputStream upload2 = AsyncStreamHelper - .toAsyncInputStream(new ClassPathResource("gridfs/another-resource.xml").getInputStream()); + Flux upload1 = DataBufferUtils.read(resource, new DefaultDataBufferFactory(), 256); + Flux upload2 = DataBufferUtils.read(new ClassPathResource("gridfs/another-resource.xml"), + new DefaultDataBufferFactory(), 256); operations.store(upload1, "foo.xml", null, null).block(); operations.store(upload2, "foo2.xml", null, null).block(); @@ -265,10 +277,11 @@ public class ReactiveGridFsTemplateTests { } @Test // DATAMONGO-1855 + @Ignore("https://jira.mongodb.org/browse/JAVARS-224") public void getResourcesByPattern() throws IOException { byte[] content = StreamUtils.copyToByteArray(resource.getInputStream()); - AsyncInputStream upload = AsyncStreamHelper.toAsyncInputStream(resource.getInputStream()); + Flux upload = DataBufferUtils.read(resource, new DefaultDataBufferFactory(), 256); operations.store(upload, "foo.xml", null, null).block(); @@ -287,6 +300,7 @@ public class ReactiveGridFsTemplateTests { } @Test // DATAMONGO-765 + @Ignore("https://jira.mongodb.org/browse/JAVARS-224") public void considersSkipLimitWhenQueryingFiles() { DataBufferFactory bufferFactory = new DefaultDataBufferFactory(); @@ -311,4 +325,57 @@ public class ReactiveGridFsTemplateTests { static class Metadata { String version; } + + @Test // + @Ignore("https://jira.mongodb.org/browse/JAVARS-224") + public void xxx() { + + GridFSBucket buckets = GridFSBuckets.create(dbFactory.getMongoDatabase()); + + DefaultDataBufferFactory factory = new DefaultDataBufferFactory(); + DefaultDataBuffer first = factory.wrap("first".getBytes()); + // DefaultDataBuffer second = factory.wrap("second".getBytes()); + + Flux source = Flux.just(first); + + // GridFSUploadPublisher objectIdGridFSUploadPublisher = buckets.uploadFromPublisher("foo.xml", + // Mono.just(ByteBuffer.wrap("hello".getBytes()))); + GridFSUploadPublisher objectIdGridFSUploadPublisher = buckets.uploadFromPublisher("foo.xml", + source.map(DataBuffer::asByteBuffer)); + Mono idPublisher = Mono.from(objectIdGridFSUploadPublisher); + idPublisher.as(StepVerifier::create).expectNextCount(1).verifyComplete(); + } + + @Test + @Ignore("https://jira.mongodb.org/browse/JAVARS-224") + public void xxx2() { + + GridFSBucket buckets = GridFSBuckets.create(dbFactory.getMongoDatabase()); + + Flux source = Flux.just(ByteBuffer.wrap("first".getBytes()), ByteBuffer.wrap("second".getBytes())); + Publisher rawSource = toPublisher(ByteBuffer.wrap("first".getBytes()), + ByteBuffer.wrap("second".getBytes())); + + // GridFSUploadPublisher objectIdGridFSUploadPublisher = buckets.uploadFromPublisher("foo.xml", + // Mono.just(ByteBuffer.wrap("hello".getBytes()))); + // GridFSUploadPublisher objectIdGridFSUploadPublisher = buckets.uploadFromPublisher("foo.xml", source); + GridFSUploadPublisher objectIdGridFSUploadPublisher = buckets.uploadFromPublisher("foo.xml", rawSource); + Mono.from(objectIdGridFSUploadPublisher).as(StepVerifier::create).expectNextCount(1).verifyComplete(); + + // idPublisher; + } + + private static Publisher toPublisher(final ByteBuffer... byteBuffers) { + return Publishers.publishAndFlatten(callback -> callback.onResult(Arrays.asList(byteBuffers), null)); + } + + private ByteBuffer hack(DataBuffer buffer) { + + ByteBuffer byteBuffer = buffer.asByteBuffer(); + ByteBuffer copy = ByteBuffer.allocate(byteBuffer.remaining()); + ByteBufferUtil.copy(byteBuffer, copy, byteBuffer.arrayOffset()); + copy.flip(); + + return copy; + } } diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/query/StringBasedMongoQueryUnitTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/query/StringBasedMongoQueryUnitTests.java index 4261c56a1..3a8a31ac8 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/query/StringBasedMongoQueryUnitTests.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/query/StringBasedMongoQueryUnitTests.java @@ -30,13 +30,15 @@ import java.util.UUID; import org.bson.BsonBinarySubType; import org.bson.Document; +import org.bson.UuidRepresentation; +import org.bson.codecs.configuration.CodecRegistry; import org.bson.types.ObjectId; import org.junit.Before; -import org.junit.Ignore; import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.Mock; import org.mockito.junit.MockitoJUnitRunner; +import org.springframework.data.mongodb.core.DbCallback; import org.springframework.data.mongodb.core.DocumentTestUtils; import org.springframework.data.mongodb.core.ExecutableFindOperation.ExecutableFind; import org.springframework.data.mongodb.core.MongoOperations; @@ -57,6 +59,9 @@ import org.springframework.data.repository.query.QueryMethodEvaluationContextPro import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.util.Base64Utils; +import com.mongodb.MongoClientSettings; +import com.mongodb.reactivestreams.client.MongoClients; + /** * Unit tests for {@link StringBasedMongoQuery}. * @@ -350,7 +355,18 @@ public class StringBasedMongoQueryUnitTests { org.springframework.data.mongodb.core.query.Query reference = new BasicQuery( "{'lastname' : { $binary:\"5PHq4zvkTYa5WbZAgvtjNg==\", $type: \"03\"}}"); - assertThat(query.getQueryObject().toJson()).isEqualTo(reference.getQueryObject().toJson()); + // CodecRegistry registry = + // MongoClientSettings.builder().uuidRepresentation(UuidRepresentation.JAVA_LEGACY).build().getCodecRegistry(); + + // TODO: use OverridableUuidRepresentationCodecRegistry instead to save resources + CodecRegistry registry = MongoClients + .create(MongoClientSettings.builder().uuidRepresentation(UuidRepresentation.JAVA_LEGACY).build()) + .getDatabase("database").getCodecRegistry(); + + // OverridableUuidRepresentationCodecRegistry + + assertThat(query.getQueryObject().toJson(registry.get(Document.class))) + .isEqualTo(reference.getQueryObject().toJson()); } @Test // DATAMONGO-2029 @@ -367,7 +383,36 @@ public class StringBasedMongoQueryUnitTests { org.springframework.data.mongodb.core.query.Query reference = new BasicQuery( "{'lastname' : { $in: [{ $binary : \"5PHq4zvkTYa5WbZAgvtjNg==\", $type : \"03\" }, { $binary : \"5PH+yjvkTYa5WbZAgvtjNg==\", $type : \"03\" }]}}"); - assertThat(query.getQueryObject().toJson()).isEqualTo(reference.getQueryObject().toJson()); + // TODO: use OverridableUuidRepresentationCodecRegistry instead to save resources + CodecRegistry registry = MongoClients + .create(MongoClientSettings.builder().uuidRepresentation(UuidRepresentation.JAVA_LEGACY).build()) + .getDatabase("database").getCodecRegistry(); + assertThat(query.getQueryObject().toJson(registry.get(Document.class))) + .isEqualTo(reference.getQueryObject().toJson()); + } + + @Test // DATAMONGO-2427 + public void shouldSupportNonQuotedUUIDCollectionReplacementWhenUsingNonLegacyUUIDCodec() { + + // TODO: use OverridableUuidRepresentationCodecRegistry instead to save resources + CodecRegistry registry = MongoClients + .create(MongoClientSettings.builder().uuidRepresentation(UuidRepresentation.STANDARD).build()) + .getDatabase("database").getCodecRegistry(); + when(operations.execute(any(DbCallback.class))).thenReturn(registry); + + UUID uuid1 = UUID.fromString("864de43b-e3ea-f1e4-3663-fb8240b659b9"); + UUID uuid2 = UUID.fromString("864de43b-cafe-f1e4-3663-fb8240b659b9"); + + ConvertingParameterAccessor accessor = StubParameterAccessor.getAccessor(converter, + (Object) Arrays.asList(uuid1, uuid2)); + StringBasedMongoQuery mongoQuery = createQueryForMethod("findByLastnameAsUUIDIn", List.class); + + org.springframework.data.mongodb.core.query.Query query = mongoQuery.createQuery(accessor); + org.springframework.data.mongodb.core.query.Query reference = new BasicQuery( + "{'lastname' : { $in: [{ $binary : \"hk3kO+Pq8eQ2Y/uCQLZZuQ==\", $type : \"04\" }, { $binary : \"hk3kO8r+8eQ2Y/uCQLZZuQ==\", $type : \"04\" }]}}"); + + assertThat(query.getQueryObject().toJson(registry.get(Document.class))) + .isEqualTo(reference.getQueryObject().toJson()); } @Test // DATAMONGO-1911 diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/test/util/MongoTestUtils.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/test/util/MongoTestUtils.java index 3924b50ab..7bb71cff8 100644 --- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/test/util/MongoTestUtils.java +++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/test/util/MongoTestUtils.java @@ -22,8 +22,10 @@ import java.time.Duration; import java.util.List; import org.bson.Document; +import org.springframework.data.mongodb.SpringDataMongoDB; import org.springframework.data.util.Version; +import com.mongodb.ConnectionString; import com.mongodb.ReadPreference; import com.mongodb.WriteConcern; import com.mongodb.client.MongoClient; @@ -55,7 +57,9 @@ public class MongoTestUtils { } public static MongoClient client(String host, int port) { - return com.mongodb.client.MongoClients.create(String.format(CONNECTION_STRING_PATTERN, host, port)); + + ConnectionString connectionString = new ConnectionString(String.format(CONNECTION_STRING_PATTERN, host, port)); + return com.mongodb.client.MongoClients.create(connectionString, SpringDataMongoDB.driverInformation()); } /** @@ -68,7 +72,9 @@ public class MongoTestUtils { } public static com.mongodb.reactivestreams.client.MongoClient reactiveClient(String host, int port) { - return MongoClients.create(String.format(CONNECTION_STRING_PATTERN, host, port)); + + ConnectionString connectionString = new ConnectionString(String.format(CONNECTION_STRING_PATTERN, host, port)); + return MongoClients.create(connectionString, SpringDataMongoDB.driverInformation()); } /** diff --git a/spring-data-mongodb/src/test/resources/org/springframework/data/mongodb/config/MongoClientNamespaceTests-context.xml b/spring-data-mongodb/src/test/resources/org/springframework/data/mongodb/config/MongoClientNamespaceTests-context.xml index 912d318ba..1bd3aa2a0 100644 --- a/spring-data-mongodb/src/test/resources/org/springframework/data/mongodb/config/MongoClientNamespaceTests-context.xml +++ b/spring-data-mongodb/src/test/resources/org/springframework/data/mongodb/config/MongoClientNamespaceTests-context.xml @@ -37,4 +37,8 @@ connection-pool-max-connection-idle-time="30" connection-pool-max-wait-time="15" /> + + + + diff --git a/src/main/asciidoc/upgrading.adoc b/src/main/asciidoc/upgrading.adoc index 507790132..5de900855 100644 --- a/src/main/asciidoc/upgrading.adoc +++ b/src/main/asciidoc/upgrading.adoc @@ -13,6 +13,9 @@ Instead of the single artifact uber-jar `mongo-java-driver`, imports are now spl * `org.mongodb:mongodb-driver-sync` (optional) * `org.mongodb:mongodb-driver-reactivestreams` (optional) +Depending on the application one of the `mongodb-driver-sync`, `mongodb-driver-reactivestreams` artifacts is is required next to the mandatory `mongodb-driver-core`. +It is possible to combine the sync and reactive drivers in one application if needed. + == Java Configuration .Java API changes @@ -140,3 +143,26 @@ Element | Comment |=== +== Other Changes + +=== UUID Types + +The MongoDB UUID representation can now be configured with different formats. +This has to be done via `MongoClientSettings` as shown in the snippet below. + +.UUid Codec Configuration +==== +[source,java] +---- + +static class Config extends AbstractMongoClientConfiguration { + + @Override + public void configureClientSettings(MongoClientSettings.Builder builder) { + builder.uuidRepresentation(UuidRepresentation.STANDARD); + } + + // ... +} +---- +====