Browse Source
Allow reuse of builders instead of resetting state after MessagePropertiesBuilder.build(). Use Java streams where possible. Slightly reorder fields to match constructor argument order. Add generics to request builders and introduce typed builder(…) methods to retain builder generics. Add builder for TailableCursorRequest. Introduce factory method on MessageListenerContainer for container creation. Change Subscription.await() to use CountDownLatch instead of polling to integrate better with ManagedBlocker. Add protected constructors to options and builder classes. Add assertions where appropriate. Move task classes into top-level types. Extract methods. Typo fixes in reference docs. Original pull request: #528.pull/528/merge
Mark Paluch
8 years ago
28 changed files with 1070 additions and 643 deletions
@ -0,0 +1,204 @@
@@ -0,0 +1,204 @@
|
||||
/* |
||||
* Copyright 2018 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 |
||||
* |
||||
* 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 org.springframework.data.mongodb.core.messaging; |
||||
|
||||
import lombok.AllArgsConstructor; |
||||
|
||||
import java.util.Arrays; |
||||
import java.util.Collections; |
||||
import java.util.HashSet; |
||||
import java.util.List; |
||||
import java.util.Optional; |
||||
import java.util.Set; |
||||
|
||||
import org.bson.BsonDocument; |
||||
import org.bson.Document; |
||||
import org.springframework.data.mongodb.core.ChangeStreamEvent; |
||||
import org.springframework.data.mongodb.core.ChangeStreamOptions; |
||||
import org.springframework.data.mongodb.core.MongoTemplate; |
||||
import org.springframework.data.mongodb.core.aggregation.Aggregation; |
||||
import org.springframework.data.mongodb.core.aggregation.AggregationOperationContext; |
||||
import org.springframework.data.mongodb.core.aggregation.PrefixingDelegatingAggregationOperationContext; |
||||
import org.springframework.data.mongodb.core.aggregation.TypeBasedAggregationOperationContext; |
||||
import org.springframework.data.mongodb.core.aggregation.TypedAggregation; |
||||
import org.springframework.data.mongodb.core.convert.MongoConverter; |
||||
import org.springframework.data.mongodb.core.convert.QueryMapper; |
||||
import org.springframework.data.mongodb.core.messaging.ChangeStreamRequest.ChangeStreamRequestOptions; |
||||
import org.springframework.data.mongodb.core.messaging.Message.MessageProperties; |
||||
import org.springframework.data.mongodb.core.messaging.SubscriptionRequest.RequestOptions; |
||||
import org.springframework.lang.Nullable; |
||||
import org.springframework.util.ClassUtils; |
||||
import org.springframework.util.ErrorHandler; |
||||
|
||||
import com.mongodb.MongoNamespace; |
||||
import com.mongodb.client.ChangeStreamIterable; |
||||
import com.mongodb.client.MongoCursor; |
||||
import com.mongodb.client.model.Collation; |
||||
import com.mongodb.client.model.changestream.ChangeStreamDocument; |
||||
import com.mongodb.client.model.changestream.FullDocument; |
||||
|
||||
/** |
||||
* {@link Task} implementation for obtaining {@link ChangeStreamDocument ChangeStreamDocuments} from MongoDB. |
||||
* |
||||
* @author Christoph Strobl |
||||
* @since 2.1 |
||||
*/ |
||||
class ChangeStreamTask extends CursorReadingTask<ChangeStreamDocument<Document>, Object> { |
||||
|
||||
private final Set<String> blacklist = new HashSet<>( |
||||
Arrays.asList("operationType", "fullDocument", "documentKey", "updateDescription", "ns")); |
||||
|
||||
private final QueryMapper queryMapper; |
||||
private final MongoConverter mongoConverter; |
||||
|
||||
@SuppressWarnings({ "unchecked", "rawtypes" }) |
||||
ChangeStreamTask(MongoTemplate template, ChangeStreamRequest<?> request, Class<?> targetType, |
||||
ErrorHandler errorHandler) { |
||||
super(template, (ChangeStreamRequest) request, (Class) targetType, errorHandler); |
||||
|
||||
queryMapper = new QueryMapper(template.getConverter()); |
||||
mongoConverter = template.getConverter(); |
||||
} |
||||
|
||||
/* |
||||
* (non-Javadoc) |
||||
* @see org.springframework.data.mongodb.core.messaging.CursorReadingTask#initCursor(org.springframework.data.mongodb.core.MongoTemplate, org.springframework.data.mongodb.core.messaging.SubscriptionRequest.RequestOptions, java.lang.Class) |
||||
*/ |
||||
@Override |
||||
protected MongoCursor<ChangeStreamDocument<Document>> initCursor(MongoTemplate template, RequestOptions options, |
||||
Class<?> targetType) { |
||||
|
||||
List<Document> filter = Collections.emptyList(); |
||||
BsonDocument resumeToken = new BsonDocument(); |
||||
Collation collation = null; |
||||
FullDocument fullDocument = FullDocument.DEFAULT; |
||||
|
||||
if (options instanceof ChangeStreamRequest.ChangeStreamRequestOptions) { |
||||
|
||||
ChangeStreamOptions changeStreamOptions = ((ChangeStreamRequestOptions) options).getChangeStreamOptions(); |
||||
filter = prepareFilter(template, changeStreamOptions); |
||||
|
||||
if (changeStreamOptions.getFilter().isPresent()) { |
||||
|
||||
Object val = changeStreamOptions.getFilter().get(); |
||||
if (val instanceof Aggregation) { |
||||
collation = ((Aggregation) val).getOptions().getCollation() |
||||
.map(org.springframework.data.mongodb.core.query.Collation::toMongoCollation).orElse(null); |
||||
} |
||||
} |
||||
|
||||
if (changeStreamOptions.getResumeToken().isPresent()) { |
||||
resumeToken = changeStreamOptions.getResumeToken().get().asDocument(); |
||||
} |
||||
|
||||
fullDocument = changeStreamOptions.getFullDocumentLookup() |
||||
.orElseGet(() -> ClassUtils.isAssignable(Document.class, targetType) ? FullDocument.DEFAULT |
||||
: FullDocument.UPDATE_LOOKUP); |
||||
} |
||||
|
||||
ChangeStreamIterable<Document> iterable = filter.isEmpty() |
||||
? template.getCollection(options.getCollectionName()).watch(Document.class) |
||||
: template.getCollection(options.getCollectionName()).watch(filter, Document.class); |
||||
|
||||
if (!resumeToken.isEmpty()) { |
||||
iterable = iterable.resumeAfter(resumeToken); |
||||
} |
||||
|
||||
if (collation != null) { |
||||
iterable = iterable.collation(collation); |
||||
} |
||||
|
||||
iterable = iterable.fullDocument(fullDocument); |
||||
|
||||
return iterable.iterator(); |
||||
} |
||||
|
||||
List<Document> prepareFilter(MongoTemplate template, ChangeStreamOptions options) { |
||||
|
||||
if (!options.getFilter().isPresent()) { |
||||
return Collections.emptyList(); |
||||
} |
||||
|
||||
Object filter = options.getFilter().get(); |
||||
if (filter instanceof Aggregation) { |
||||
Aggregation agg = (Aggregation) filter; |
||||
AggregationOperationContext context = agg instanceof TypedAggregation |
||||
? new TypeBasedAggregationOperationContext(((TypedAggregation<?>) agg).getInputType(), |
||||
template.getConverter().getMappingContext(), queryMapper) |
||||
: Aggregation.DEFAULT_CONTEXT; |
||||
|
||||
return agg.toPipeline(new PrefixingDelegatingAggregationOperationContext(context, "fullDocument", blacklist)); |
||||
} else if (filter instanceof List) { |
||||
return (List<Document>) filter; |
||||
} else { |
||||
throw new IllegalArgumentException( |
||||
"ChangeStreamRequestOptions.filter mut be either an Aggregation or a plain list of Documents"); |
||||
} |
||||
} |
||||
|
||||
@Override |
||||
protected Message<ChangeStreamDocument<Document>, Object> createMessage(ChangeStreamDocument<Document> source, |
||||
Class<Object> targetType, RequestOptions options) { |
||||
|
||||
// namespace might be null for eg. OperationType.INVALIDATE
|
||||
MongoNamespace namespace = Optional.ofNullable(source.getNamespace()) |
||||
.orElse(new MongoNamespace("unknown", options.getCollectionName())); |
||||
|
||||
return new ChangeStreamEventMessage<>(new ChangeStreamEvent<>(source, targetType, mongoConverter), MessageProperties |
||||
.builder().databaseName(namespace.getDatabaseName()).collectionName(namespace.getCollectionName()).build()); |
||||
} |
||||
|
||||
/** |
||||
* {@link Message} implementation for ChangeStreams |
||||
* |
||||
* @since 2.1 |
||||
*/ |
||||
@AllArgsConstructor |
||||
static class ChangeStreamEventMessage<T> implements Message<ChangeStreamDocument<Document>, T> { |
||||
|
||||
private final ChangeStreamEvent<T> delegate; |
||||
private final MessageProperties messageProperties; |
||||
|
||||
/* |
||||
* (non-Javadoc) |
||||
* @see org.springframework.data.mongodb.core.messaging.Message#getRaw() |
||||
*/ |
||||
@Nullable |
||||
@Override |
||||
public ChangeStreamDocument<Document> getRaw() { |
||||
return delegate.getRaw(); |
||||
} |
||||
|
||||
/* |
||||
* (non-Javadoc) |
||||
* @see org.springframework.data.mongodb.core.messaging.Message#getBody() |
||||
*/ |
||||
@Nullable |
||||
@Override |
||||
public T getBody() { |
||||
return delegate.getBody(); |
||||
} |
||||
|
||||
/* |
||||
* (non-Javadoc) |
||||
* @see org.springframework.data.mongodb.core.messaging.Message#getProperties() |
||||
*/ |
||||
@Override |
||||
public MessageProperties getProperties() { |
||||
return this.messageProperties; |
||||
} |
||||
} |
||||
} |
||||
@ -0,0 +1,251 @@
@@ -0,0 +1,251 @@
|
||||
/* |
||||
* Copyright 2018 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 |
||||
* |
||||
* 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 org.springframework.data.mongodb.core.messaging; |
||||
|
||||
import java.time.Duration; |
||||
import java.util.concurrent.CountDownLatch; |
||||
import java.util.concurrent.TimeUnit; |
||||
|
||||
import org.springframework.dao.DataAccessResourceFailureException; |
||||
import org.springframework.data.mongodb.core.MongoTemplate; |
||||
import org.springframework.data.mongodb.core.messaging.Message.MessageProperties; |
||||
import org.springframework.data.mongodb.core.messaging.SubscriptionRequest.RequestOptions; |
||||
import org.springframework.lang.Nullable; |
||||
import org.springframework.util.ErrorHandler; |
||||
|
||||
import com.mongodb.client.MongoCursor; |
||||
import com.mysema.commons.lang.Assert; |
||||
|
||||
/** |
||||
* @author Christoph Strobl |
||||
* @author Mark Paluch |
||||
* @param <T> type of objects returned by the cursor. |
||||
* @param <R> conversion target type. |
||||
* @since 2.1 |
||||
*/ |
||||
abstract class CursorReadingTask<T, R> implements Task { |
||||
|
||||
private final Object lifecycleMonitor = new Object(); |
||||
|
||||
private final MongoTemplate template; |
||||
private final SubscriptionRequest<T, R, RequestOptions> request; |
||||
private final Class<R> targetType; |
||||
private final ErrorHandler errorHandler; |
||||
private final CountDownLatch awaitStart = new CountDownLatch(1); |
||||
|
||||
private State state = State.CREATED; |
||||
|
||||
private MongoCursor<T> cursor; |
||||
|
||||
/** |
||||
* @param template must not be {@literal null}. |
||||
* @param request must not be {@literal null}. |
||||
* @param targetType must not be {@literal null}. |
||||
*/ |
||||
@SuppressWarnings({ "unchecked", "rawtypes" }) |
||||
CursorReadingTask(MongoTemplate template, SubscriptionRequest<?, ? super T, ? extends RequestOptions> request, |
||||
Class<R> targetType, ErrorHandler errorHandler) { |
||||
|
||||
this.template = template; |
||||
this.request = (SubscriptionRequest) request; |
||||
this.targetType = targetType; |
||||
this.errorHandler = errorHandler; |
||||
} |
||||
|
||||
/* |
||||
* (non-Javadoc) |
||||
* @see java.lang.Runnable |
||||
*/ |
||||
@Override |
||||
public void run() { |
||||
|
||||
start(); |
||||
|
||||
while (isRunning()) { |
||||
try { |
||||
T next = getNext(); |
||||
if (next != null) { |
||||
emitMessage(createMessage(next, targetType, request.getRequestOptions())); |
||||
} else { |
||||
Thread.sleep(10); |
||||
} |
||||
} catch (InterruptedException e) { |
||||
|
||||
synchronized (lifecycleMonitor) { |
||||
state = State.CANCELLED; |
||||
} |
||||
Thread.interrupted(); |
||||
} catch (RuntimeException e) { |
||||
|
||||
Exception translated = template.getExceptionTranslator().translateExceptionIfPossible(e); |
||||
Exception toHandle = translated != null ? translated : e; |
||||
|
||||
errorHandler.handleError(toHandle); |
||||
} |
||||
} |
||||
} |
||||
|
||||
/** |
||||
* Initialize the Task by 1st setting the current state to {@link State#STARTING starting} indicating the |
||||
* initialization procedure. <br /> |
||||
* Moving on the underlying {@link MongoCursor} gets {@link #initCursor(MongoTemplate, RequestOptions) created} and is |
||||
* {@link #isValidCursor(MongoCursor) health checked}. Once a valid {@link MongoCursor} is created the {@link #state} |
||||
* is set to {@link State#RUNNING running}. If the health check is not passed the {@link MongoCursor} is immediately |
||||
* {@link MongoCursor#close() closed} and a new {@link MongoCursor} is requested until a valid one is retrieved or the |
||||
* {@link #state} changes. |
||||
*/ |
||||
private void start() { |
||||
|
||||
synchronized (lifecycleMonitor) { |
||||
if (!State.RUNNING.equals(state)) { |
||||
state = State.STARTING; |
||||
} |
||||
} |
||||
|
||||
do { |
||||
|
||||
boolean valid = false; |
||||
|
||||
synchronized (lifecycleMonitor) { |
||||
|
||||
if (State.STARTING.equals(state)) { |
||||
|
||||
MongoCursor<T> cursor = initCursor(template, request.getRequestOptions(), targetType); |
||||
valid = isValidCursor(cursor); |
||||
if (valid) { |
||||
this.cursor = cursor; |
||||
state = State.RUNNING; |
||||
} else { |
||||
cursor.close(); |
||||
} |
||||
} |
||||
} |
||||
|
||||
if (!valid) { |
||||
|
||||
try { |
||||
Thread.sleep(100); |
||||
} catch (InterruptedException e) { |
||||
|
||||
synchronized (lifecycleMonitor) { |
||||
state = State.CANCELLED; |
||||
} |
||||
Thread.interrupted(); |
||||
} |
||||
} |
||||
} while (State.STARTING.equals(getState())); |
||||
|
||||
if (awaitStart.getCount() == 1) { |
||||
awaitStart.countDown(); |
||||
} |
||||
} |
||||
|
||||
protected abstract MongoCursor<T> initCursor(MongoTemplate template, RequestOptions options, Class<?> targetType); |
||||
|
||||
/* |
||||
* (non-Javadoc) |
||||
* @see org.springframework.data.mongodb.core.messaging.Cancelable#cancel() |
||||
*/ |
||||
@Override |
||||
public void cancel() throws DataAccessResourceFailureException { |
||||
|
||||
synchronized (lifecycleMonitor) { |
||||
|
||||
if (State.RUNNING.equals(state) || State.STARTING.equals(state)) { |
||||
this.state = State.CANCELLED; |
||||
if (cursor != null) { |
||||
cursor.close(); |
||||
} |
||||
} |
||||
} |
||||
} |
||||
|
||||
/* |
||||
* (non-Javadoc) |
||||
* @see org.springframework.scheduling.SchedulingAwareRunnable#isLongLived() |
||||
*/ |
||||
@Override |
||||
public boolean isLongLived() { |
||||
return true; |
||||
} |
||||
|
||||
/* |
||||
* (non-Javadoc) |
||||
* @see org.springframework.data.mongodb.core.messaging.Task#getState() |
||||
*/ |
||||
@Override |
||||
public State getState() { |
||||
|
||||
synchronized (lifecycleMonitor) { |
||||
return state; |
||||
} |
||||
} |
||||
|
||||
/* |
||||
* (non-Javadoc) |
||||
* @see org.springframework.data.mongodb.core.messaging.Task#awaitStart(java.time.Duration) |
||||
*/ |
||||
@Override |
||||
public boolean awaitStart(Duration timeout) throws InterruptedException { |
||||
|
||||
Assert.notNull(timeout, "Timeout must not be null!"); |
||||
Assert.isFalse(timeout.isNegative(), "Timeout must not be negative!"); |
||||
|
||||
return awaitStart.await(timeout.toNanos(), TimeUnit.NANOSECONDS); |
||||
} |
||||
|
||||
protected Message<T, R> createMessage(T source, Class<R> targetType, RequestOptions options) { |
||||
|
||||
SimpleMessage<T, T> message = new SimpleMessage<>(source, source, MessageProperties.builder() |
||||
.databaseName(template.getDb().getName()).collectionName(options.getCollectionName()).build()); |
||||
|
||||
return new LazyMappingDelegatingMessage<>(message, targetType, template.getConverter()); |
||||
} |
||||
|
||||
private boolean isRunning() { |
||||
return State.RUNNING.equals(getState()); |
||||
} |
||||
|
||||
@SuppressWarnings("unchecked") |
||||
private void emitMessage(Message<T, R> message) { |
||||
request.getMessageListener().onMessage((Message) message); |
||||
} |
||||
|
||||
@Nullable |
||||
private T getNext() { |
||||
|
||||
synchronized (lifecycleMonitor) { |
||||
if (State.RUNNING.equals(state)) { |
||||
return cursor.tryNext(); |
||||
} |
||||
} |
||||
|
||||
throw new IllegalStateException(String.format("Cursor %s is not longer open.", cursor)); |
||||
} |
||||
|
||||
private static boolean isValidCursor(@Nullable MongoCursor<?> cursor) { |
||||
|
||||
if (cursor == null) { |
||||
return false; |
||||
} |
||||
|
||||
if (cursor.getServerCursor() == null || cursor.getServerCursor().getId() == 0) { |
||||
return false; |
||||
} |
||||
|
||||
return true; |
||||
} |
||||
} |
||||
@ -0,0 +1,85 @@
@@ -0,0 +1,85 @@
|
||||
/* |
||||
* Copyright 2018 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 |
||||
* |
||||
* 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 org.springframework.data.mongodb.core.messaging; |
||||
|
||||
import lombok.ToString; |
||||
|
||||
import org.bson.Document; |
||||
import org.springframework.data.mongodb.core.convert.MongoConverter; |
||||
import org.springframework.util.ClassUtils; |
||||
|
||||
/** |
||||
* @author Christoph Strobl |
||||
* @author Mark Paluch |
||||
* @since 2.1 |
||||
*/ |
||||
@ToString(of = { "delegate", "targetType" }) |
||||
class LazyMappingDelegatingMessage<S, T> implements Message<S, T> { |
||||
|
||||
private final Message<S, ?> delegate; |
||||
private final Class<T> targetType; |
||||
private final MongoConverter converter; |
||||
|
||||
LazyMappingDelegatingMessage(Message<S, ?> delegate, Class<T> targetType, MongoConverter converter) { |
||||
|
||||
this.delegate = delegate; |
||||
this.targetType = targetType; |
||||
this.converter = converter; |
||||
} |
||||
|
||||
/* |
||||
* (non-Javadoc) |
||||
* @see org.springframework.data.mongodb.core.messaging.Message#getRaw() |
||||
*/ |
||||
@Override |
||||
public S getRaw() { |
||||
return delegate.getRaw(); |
||||
} |
||||
|
||||
/* |
||||
* (non-Javadoc) |
||||
* @see org.springframework.data.mongodb.core.messaging.Message#getBody() |
||||
*/ |
||||
@Override |
||||
public T getBody() { |
||||
|
||||
if (delegate.getBody() == null || targetType.equals(delegate.getBody().getClass())) { |
||||
return targetType.cast(delegate.getBody()); |
||||
} |
||||
|
||||
Object messageBody = delegate.getBody(); |
||||
|
||||
if (ClassUtils.isAssignable(Document.class, messageBody.getClass())) { |
||||
return converter.read(targetType, (Document) messageBody); |
||||
} |
||||
|
||||
if (converter.getConversionService().canConvert(messageBody.getClass(), targetType)) { |
||||
return converter.getConversionService().convert(messageBody, targetType); |
||||
} |
||||
|
||||
throw new IllegalArgumentException( |
||||
String.format("No converter found capable of converting %s to %s", messageBody.getClass(), targetType)); |
||||
} |
||||
|
||||
/* |
||||
* (non-Javadoc) |
||||
* @see org.springframework.data.mongodb.core.messaging.Message#getProperties() |
||||
*/ |
||||
@Override |
||||
public MessageProperties getProperties() { |
||||
return delegate.getProperties(); |
||||
} |
||||
} |
||||
@ -0,0 +1,80 @@
@@ -0,0 +1,80 @@
|
||||
/* |
||||
* Copyright 2018 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 |
||||
* |
||||
* 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 org.springframework.data.mongodb.core.messaging; |
||||
|
||||
import org.bson.Document; |
||||
import org.springframework.data.mongodb.core.MongoTemplate; |
||||
import org.springframework.data.mongodb.core.convert.QueryMapper; |
||||
import org.springframework.data.mongodb.core.messaging.SubscriptionRequest.RequestOptions; |
||||
import org.springframework.data.mongodb.core.messaging.TailableCursorRequest.TailableCursorRequestOptions; |
||||
import org.springframework.data.mongodb.core.query.Query; |
||||
import org.springframework.util.ErrorHandler; |
||||
|
||||
import com.mongodb.CursorType; |
||||
import com.mongodb.client.FindIterable; |
||||
import com.mongodb.client.MongoCursor; |
||||
import com.mongodb.client.model.Collation; |
||||
|
||||
/** |
||||
* @author Christoph Strobl |
||||
* @since 2.1 |
||||
*/ |
||||
class TailableCursorTask extends CursorReadingTask<Document, Object> { |
||||
|
||||
private QueryMapper queryMapper; |
||||
|
||||
@SuppressWarnings({ "unchecked", "rawtypes" }) |
||||
public TailableCursorTask(MongoTemplate template, TailableCursorRequest<?> request, Class<?> targetType, |
||||
ErrorHandler errorHandler) { |
||||
super(template, (TailableCursorRequest) request, (Class) targetType, errorHandler); |
||||
queryMapper = new QueryMapper(template.getConverter()); |
||||
} |
||||
|
||||
/* |
||||
* (non-Javadoc) |
||||
* @see org.springframework.data.mongodb.core.messaging.CursorReadingTask#initCursor(org.springframework.data.mongodb.core.MongoTemplate, org.springframework.data.mongodb.core.messaging.SubscriptionRequest.RequestOptions, java.lang.Class) |
||||
*/ |
||||
@Override |
||||
protected MongoCursor<Document> initCursor(MongoTemplate template, RequestOptions options, Class<?> targetType) { |
||||
|
||||
Document filter = new Document(); |
||||
Collation collation = null; |
||||
|
||||
if (options instanceof TailableCursorRequest.TailableCursorRequestOptions) { |
||||
|
||||
TailableCursorRequestOptions requestOptions = (TailableCursorRequestOptions) options; |
||||
if (requestOptions.getQuery().isPresent()) { |
||||
|
||||
Query query = requestOptions.getQuery().get(); |
||||
|
||||
filter.putAll(queryMapper.getMappedObject(query.getQueryObject(), template.getConverter().getMappingContext() |
||||
.getPersistentEntity(targetType.equals(Document.class) ? Object.class : targetType))); |
||||
|
||||
collation = query.getCollation().map(org.springframework.data.mongodb.core.query.Collation::toMongoCollation) |
||||
.orElse(null); |
||||
} |
||||
} |
||||
|
||||
FindIterable<Document> iterable = template.getCollection(options.getCollectionName()).find(filter) |
||||
.cursorType(CursorType.TailableAwait).noCursorTimeout(true); |
||||
|
||||
if (collation != null) { |
||||
iterable = iterable.collation(collation); |
||||
} |
||||
|
||||
return iterable.iterator(); |
||||
} |
||||
} |
||||
@ -0,0 +1,45 @@
@@ -0,0 +1,45 @@
|
||||
/* |
||||
* Copyright 2018 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 |
||||
* |
||||
* 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 org.springframework.data.mongodb.core.messaging; |
||||
|
||||
import static org.assertj.core.api.Assertions.*; |
||||
import static org.springframework.data.mongodb.core.query.Criteria.*; |
||||
|
||||
import org.bson.Document; |
||||
import org.junit.Test; |
||||
import org.springframework.data.mongodb.core.messaging.DefaultMessageListenerContainerTests.Person; |
||||
import org.springframework.data.mongodb.core.query.Query; |
||||
|
||||
/** |
||||
* Unit tests for {@link TailableCursorRequest}. |
||||
* |
||||
* @author Mark Paluch |
||||
*/ |
||||
public class TailableCursorRequestUnitTests { |
||||
|
||||
@Test // DATAMONGO-1803
|
||||
public void shouldBuildRequest() { |
||||
|
||||
MessageListener<Document, Person> listener = System.out::println; |
||||
|
||||
TailableCursorRequest<Person> request = TailableCursorRequest.builder(listener).collection("foo") |
||||
.filter(Query.query(where("firstname").is("bar"))).build(); |
||||
|
||||
assertThat(request.getRequestOptions().getCollectionName()).isEqualTo("foo"); |
||||
assertThat(request.getRequestOptions().getQuery()).isPresent(); |
||||
assertThat(request.getMessageListener()).isEqualTo(listener); |
||||
} |
||||
} |
||||
Loading…
Reference in new issue