tagging release 3.272

Author: dxcity

ReleaseNotes.txt

@@ -1,4 +1,9 @@
 
+QDS 3.272:
+
+* [QD-1125] dxFeed webservice: Use session filter for one-time REST requests
+  - not a public API: for internal use only
+
 QDS 3.271:
 
 * [QD-1121] dxFeed webservice: Add support for security checks

auth/pom.xml

@@ -14,7 +14,7 @@
     <parent>
         <artifactId>QD</artifactId>
         <groupId>com.devexperts.qd</groupId>
-        <version>3.271</version>
+        <version>3.272</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

dxfeed-api/pom.xml

@@ -14,7 +14,7 @@
 	<parent>
 		<artifactId>QD</artifactId>
 		<groupId>com.devexperts.qd</groupId>
-		<version>3.271</version>
+		<version>3.272</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<modelVersion>4.0.0</modelVersion>

dxfeed-bin/pom.xml

@@ -14,7 +14,7 @@
 	<parent>
 		<artifactId>QD</artifactId>
 		<groupId>com.devexperts.qd</groupId>
-		<version>3.271</version>
+		<version>3.272</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<modelVersion>4.0.0</modelVersion>

dxfeed-codegen-verify/pom.xml

@@ -14,7 +14,7 @@
 	<parent>
 		<artifactId>QD</artifactId>
 		<groupId>com.devexperts.qd</groupId>
-		<version>3.271</version>
+		<version>3.272</version>
 	</parent>
 	<modelVersion>4.0.0</modelVersion>
 

dxfeed-codegen/pom.xml

@@ -14,7 +14,7 @@
 	<parent>
 		<artifactId>QD</artifactId>
 		<groupId>com.devexperts.qd</groupId>
-		<version>3.271</version>
+		<version>3.272</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 	<modelVersion>4.0.0</modelVersion>

dxfeed-impl/pom.xml

@@ -14,7 +14,7 @@
 	<parent>
 		<artifactId>QD</artifactId>
 		<groupId>com.devexperts.qd</groupId>
-		<version>3.271</version>
+		<version>3.272</version>
 	</parent>
 	<modelVersion>4.0.0</modelVersion>
 

dxfeed-impl/src/main/java/com/dxfeed/api/impl/DXFeedImpl.java

@@ -42,6 +42,7 @@ public class DXFeedImpl extends DXFeed {
 
 	private final DXEndpointImpl endpoint;
 	private final QDFilter filter;
+	private final QDFilter.UpdateListener filterListener;
 	private final RecordMode retrieveMode;
 	private final QDAgent.Builder[] eventProcessorAgentBuilders = new QDAgent.Builder[N_CONTRACTS];
 	private final IndexedSet<DXFeedSubscription<?>, EventProcessor<?, ?>> eventProcessors =
@@ -67,6 +68,13 @@ public DXFeedImpl(DXEndpointImpl endpoint) {
 	public DXFeedImpl(DXEndpointImpl endpoint, QDFilter filter) {
 		this.endpoint = endpoint;
 		this.filter = filter;
+		if (filter.isDynamic()) {
+			// Add filter to force automatic update
+			filterListener = this::updateSubscriptionsOnFilterUpdate;
+			filter.getUpdated().addUpdateListener(filterListener);
+		} else {
+			filterListener = null;
+		}
 
 		RecordMode mode = RecordMode.FLAGGED_DATA.withAttachment();
 		if (endpoint.getQDEndpoint().hasEventTimeSequence())
@@ -102,11 +110,13 @@ public static void clearDataInBuffer(RecordBuffer buf, boolean keepTime) {
 	}
 
 	public void awaitTerminationAndCloseImpl() throws InterruptedException {
+		// Async iteration
 		for (EventProcessor<?, ?> processor : eventProcessors.toArray(new EventProcessor[eventProcessors.size()])) {
 			if (processor == null)
 				break;
-			processor.awaitTerminationAndClose();
+			processor.awaitTermination();
 		}
+		closeImpl();
 	}
 
 	public void closeImpl() {
@@ -119,17 +129,23 @@ public void closeImpl() {
 			// no need to sync, because nothing is added or removed when endpoint is closed
 			closeComponents();
 		}
-		eventProcessors.clear();
-		closeables.clear();
 	}
 
 	private void closeComponents() {
 		for (EventProcessor<?, ?> processor : eventProcessors)
 			processor.close(false);
+		eventProcessors.clear();
 		for (Closeable c : closeables)
 			c.close();
+		closeables.clear();
 		if (lastEventsProcessor != null)
 			lastEventsProcessor.close();
+
+		// Remove listener for dynamic filter
+		if (filterListener != null) {
+			assert filter.isDynamic(); // filter listener is created only for dynamic filters
+			filter.getUpdated().removeUpdateListener(filterListener);
+		}
 	}
 
 	private void removeEventProcessor(DXFeedSubscription<?> subscription) {
@@ -153,18 +169,21 @@ private void removeCloseable(Closeable c) {
 	@Override
 	@SuppressWarnings("unchecked")
 	public void attachSubscription(DXFeedSubscription<?> subscription) {
+		//FIXME Store listener for later removal on detach. Check for double addition.
 		subscription.addChangeListener(new SubscriptionChangeListener(subscription, false));
 	}
 
 	@Override
 	@SuppressWarnings("unchecked")
 	public void detachSubscription(DXFeedSubscription<?> subscription) {
+		//FIXME Remove the same listener that was added before.
 		subscription.removeChangeListener(new SubscriptionChangeListener(subscription, false));
 	}
 
 	@Override
 	@SuppressWarnings("unchecked")
 	public void detachSubscriptionAndClear(DXFeedSubscription<?> subscription) {
+		//FIXME Remove the same listener that was added before.
 		subscription.removeChangeListener(new SubscriptionChangeListener(subscription, true));
 	}
 
@@ -179,7 +198,7 @@ public <E extends LastingEvent<?>> E getLastEvent(E event) {
 		int cipher = endpoint.encode(qdSymbol);
 
 		// check if symbol is accepted by filter
-		if (!filter.accept(QDContract.TICKER, delegate.getRecord(), cipher, qdSymbol))
+		if (!filter.getUpdatedFilter().accept(QDContract.TICKER, delegate.getRecord(), cipher, qdSymbol))
 			return event;
 		LocalAddBatch lb = getLocalAddBatch();
 		if (ticker.getDataIfAvailable(lb.owner, delegate.getRecord(), cipher, qdSymbol))
@@ -198,7 +217,7 @@ public <E extends LastingEvent<?>> E getLastEventIfSubscribed(Class<E> eventType
 		int cipher = endpoint.encode(qdSymbol);
 
 		// check if symbol is accepted by filter
-		if (!filter.accept(QDContract.TICKER, delegate.getRecord(), cipher, qdSymbol))
+		if (!filter.getUpdatedFilter().accept(QDContract.TICKER, delegate.getRecord(), cipher, qdSymbol))
 			return null;
 		LocalAddBatch lb = getLocalAddBatch();
 		if (lastEventsProcessor.ticker.getDataIfSubscribed(lb.owner, delegate.getRecord(), cipher, qdSymbol))
@@ -219,7 +238,7 @@ public <E extends LastingEvent<?>> Promise<E> getLastEventPromise(Class<E> event
 		int cipher = endpoint.encode(qdSymbol);
 
 		// check if symbol is accepted by filter
-		if (!filter.accept(QDContract.TICKER, delegate.getRecord(), cipher, qdSymbol))
+		if (!filter.getUpdatedFilter().accept(QDContract.TICKER, delegate.getRecord(), cipher, qdSymbol))
 			return Promise.failed(new CancellationException("cancel"));
 		// optimization for single event -- check that it is immediately available without subscription
 		LocalAddBatch lb = getLocalAddBatch();
@@ -254,7 +273,7 @@ public <E extends LastingEvent<?>> List<Promise<E>> getLastEventsPromises(Class<
 			int cipher = endpoint.encode(qdSymbol);
 
 			// check if symbol is accepted by filter
-			if (!filter.accept(QDContract.TICKER, delegate.getRecord(), cipher, qdSymbol)) {
+			if (!filter.getUpdatedFilter().accept(QDContract.TICKER, delegate.getRecord(), cipher, qdSymbol)) {
 				result.add(Promise.failed(new CancellationException("cancel")));
 				continue;
 			}
@@ -400,7 +419,7 @@ private <E extends IndexedEvent<?>> List<E> fetchFromHistoryIfSubscribed(Object
 		int cipher = endpoint.encode(qdSymbol);
 
 		// check if symbol is accepted by filter
-		if (!filter.accept(QDContract.HISTORY, delegate.getRecord(), cipher, qdSymbol))
+		if (!filter.getUpdatedFilter().accept(QDContract.HISTORY, delegate.getRecord(), cipher, qdSymbol))
 			return Collections.emptyList();
 		// check subscription
 		if (!history.isSubscribed(delegate.getRecord(), cipher, qdSymbol, fromQDTime))
@@ -419,7 +438,7 @@ private <E extends IndexedEvent<?>> Promise<List<E>> fetchOrSubscribeFromHistory
 		int cipher = endpoint.encode(qdSymbol);
 
 		// check if symbol is accepted by filter
-		if (!filter.accept(QDContract.HISTORY, delegate.getRecord(), cipher, qdSymbol))
+		if (!filter.getUpdatedFilter().accept(QDContract.HISTORY, delegate.getRecord(), cipher, qdSymbol))
 			return Promise.failed(new CancellationException("cancel"));
 		// check if data is available in history without subscription
 		HistoryFetchResult<E> fetch = new HistoryFetchResult<>(symbol, fromQDTime, delegate, true);
@@ -524,6 +543,19 @@ public long getAggregationPeriodMillis() {
 		return aggregationPeriodMillis;
 	}
 
+	private void updateSubscriptionsOnFilterUpdate(QDFilter updatedFilter) {
+		// Filter parameter is ignored since agents will always use latest filter version on subscription
+		for (EventProcessor<?, ?> processor : eventProcessors) {
+			Set<?> symbols = processor.subscription.getSymbols();
+			EnumMap<QDContract, RecordBuffer> sub = toSubscription(processor.subscription, symbols, true);
+			for (QDContract contract : sub.keySet()) {
+				RecordBuffer buffer = sub.get(contract);
+				processor.getOrCreateAgent(contract).setSubscription(buffer);
+				buffer.release();
+			}
+		}
+	}
+
 	private <E> void executePromiseHandler(final Promise<E> promise, final PromiseHandler<? super E> handler) {
 		if (handler != null)
 			endpoint.getOrCreateExecutor().execute(() -> handler.promiseDone(promise));
@@ -884,7 +916,7 @@ QDAgent getOrCreateAgent(QDContract contract) {
 			QDAgent agent = agents[contract.ordinal()];
 			if (agent != null)
 				return agent;
-			agent = eventProcessorAgentBuilders[contract.ordinal()].build();
+			agent = eventProcessorAgentBuilders[contract.ordinal()].withFilter(filter.getUpdatedFilter()).build();
 			if (endpoint.getRole() == DXEndpoint.Role.STREAM_FEED)
 				agent.setBufferOverflowStrategy(QDAgent.BufferOverflowStrategy.BLOCK);
 			agents[contract.ordinal()] = agent;
@@ -942,11 +974,6 @@ private void signalNoMoreDataToProcess() {
 				terminationLatch.countDown();
 		}
 
-		void awaitTerminationAndClose() throws InterruptedException {
-			awaitTermination();
-			close(false);
-		}
-
 		private void awaitTermination() throws InterruptedException {
 			terminationLatch = new CountDownLatch(1);
 			if (hasMoreDataToProcess()) {