tagging release 3.271

Author: dxcity

ReleaseNotes.txt

@@ -1,4 +1,12 @@
 
+QDS 3.271:
+
+* [QD-1121] dxFeed webservice: Add support for security checks
+  - not a public API: for internal use only
+* [QD-984] Improve DebugDumpReader to see HistoryBuffer state
+  - A bug with wrong History "sub" information due to non-resolved enums is also fixed
+* [QD-988] BUG FIX: Monitoring tests sometimes fall because the ports are already in use
+
 QDS 3.270:
 
 * [QD-1123] dxFeed API: Make CandleSymbol attribute classes Serializable

auth/pom.xml

@@ -14,7 +14,7 @@
     <parent>
         <artifactId>QD</artifactId>
         <groupId>com.devexperts.qd</groupId>
-        <version>3.270</version>
+        <version>3.271</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.270</version>
+		<version>3.271</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.270</version>
+		<version>3.271</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.270</version>
+		<version>3.271</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.270</version>
+		<version>3.271</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.270</version>
+		<version>3.271</version>
 	</parent>
 	<modelVersion>4.0.0</modelVersion>
 

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

@@ -41,6 +41,7 @@ public class DXFeedImpl extends DXFeed {
 	private static final ThreadLocal<LocalRemoveBatch> LOCAL_REMOVE_BATCH = new ThreadLocal<>();
 
 	private final DXEndpointImpl endpoint;
+	private final QDFilter filter;
 	private final RecordMode retrieveMode;
 	private final QDAgent.Builder[] eventProcessorAgentBuilders = new QDAgent.Builder[N_CONTRACTS];
 	private final IndexedSet<DXFeedSubscription<?>, EventProcessor<?, ?>> eventProcessors =
@@ -49,23 +50,40 @@ public class DXFeedImpl extends DXFeed {
 	private final LastEventsProcessor lastEventsProcessor; // != null when we have QDTicker in endpoint
 	private final long aggregationPeriodMillis;
 
-	DXFeedImpl(DXEndpointImpl endpoint) {
+	// Default constructor for DXEndpointImpl
+	@SuppressWarnings("deprecation")
+	public DXFeedImpl(DXEndpointImpl endpoint) {
+		this(endpoint, QDFilter.ANYTHING);
+	}
+
+	/**
+	 * Allows to create DXFeed instance with the specific filter from the given endpoint. Endpoint will not handle
+	 * this feed's lifecycle - call closeImpl() or awaitTerminationAndCloseImpl() to clean up resources!
+	 * @param endpoint
+	 * @param filter
+	 * @deprecated For internal use only, do not use!
+	 */
+	@Deprecated
+	public DXFeedImpl(DXEndpointImpl endpoint, QDFilter filter) {
 		this.endpoint = endpoint;
+		this.filter = filter;
+
 		RecordMode mode = RecordMode.FLAGGED_DATA.withAttachment();
 		if (endpoint.getQDEndpoint().hasEventTimeSequence())
 			mode = mode.withEventTimeSequence();
 		retrieveMode = mode;
 		for (QDContract contract : endpoint.getContracts()) {
 			QDAgent.Builder builder = endpoint.getCollector(contract).agentBuilder()
 				.withHistorySnapshot(true)
+				.withFilter(filter)
 				.withAttachmentStrategy(EVENT_PROCESSOR_ATTACHMENT_STRATEGY);
 			eventProcessorAgentBuilders[contract.ordinal()] = builder;
 		}
 		QDTicker ticker = (QDTicker) endpoint.getCollector(QDContract.TICKER);
 		if (ticker == null) {
 			lastEventsProcessor = null;
 		} else {
-			lastEventsProcessor = new LastEventsProcessor(ticker);
+			lastEventsProcessor = new LastEventsProcessor(ticker, filter);
 			lastEventsProcessor.start();
 		}
 		aggregationPeriodMillis = endpoint.hasProperty(DXEndpoint.DXFEED_AGGREGATION_PERIOD_PROPERTY) ?
@@ -92,15 +110,24 @@ public void awaitTerminationAndCloseImpl() throws InterruptedException {
 	}
 
 	public void closeImpl() {
-		assert endpoint.isClosed(); // assert that close impl is called on endpoint that is already marked as closed
-		// no need to sync, because nothing is added or removed when endpoint is closed
+		// cannot be sure that endpoint is already closed
+		if (!endpoint.isClosed()) {
+			synchronized (endpoint.getLock()) {
+				closeComponents();
+			}
+		} else {
+			// 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();
-		// no need to sync, because nothing is added or removed when endpoint is closed
 		for (Closeable c : closeables)
 			c.close();
-		closeables.clear();
 		if (lastEventsProcessor != null)
 			lastEventsProcessor.close();
 	}
@@ -148,9 +175,13 @@ public <E extends LastingEvent<?>> E getLastEvent(E event) {
 		if (delegate == null)
 			return event;
 		QDTicker ticker = (QDTicker)endpoint.getCollector(QDContract.TICKER); // assert != null (if we have delegate)
-		LocalAddBatch lb = getLocalAddBatch();
 		String qdSymbol = delegate.getQDSymbolByEvent(event);
 		int cipher = endpoint.encode(qdSymbol);
+
+		// check if symbol is accepted by filter
+		if (!filter.accept(QDContract.TICKER, delegate.getRecord(), cipher, qdSymbol))
+			return event;
+		LocalAddBatch lb = getLocalAddBatch();
 		if (ticker.getDataIfAvailable(lb.owner, delegate.getRecord(), cipher, qdSymbol))
 			return delegate.getEvent(event, lb.owner.cursor());
 		// not found -- return unmodified event
@@ -163,9 +194,13 @@ public <E extends LastingEvent<?>> E getLastEventIfSubscribed(Class<E> eventType
 		if (delegate == null)
 			return null;
 		assert lastEventsProcessor != null; // if we have delegate
-		LocalAddBatch lb = getLocalAddBatch();
 		String qdSymbol = delegate.getQDSymbolByEventSymbol(symbol);
 		int cipher = endpoint.encode(qdSymbol);
+
+		// check if symbol is accepted by filter
+		if (!filter.accept(QDContract.TICKER, delegate.getRecord(), cipher, qdSymbol))
+			return null;
+		LocalAddBatch lb = getLocalAddBatch();
 		if (lastEventsProcessor.ticker.getDataIfSubscribed(lb.owner, delegate.getRecord(), cipher, qdSymbol))
 			return delegate.createEvent(symbol, lb.owner.cursor());
 		// not subscribed -- return null
@@ -180,10 +215,14 @@ public <E extends LastingEvent<?>> Promise<E> getLastEventPromise(Class<E> event
 		if (delegate == null)
 			return Promise.failed(new IllegalArgumentException(INVALID_EVENT_MSG));
 		assert lastEventsProcessor != null; // if we have delegate
-		LocalAddBatch lb = getLocalAddBatch();
 		String qdSymbol = delegate.getQDSymbolByEventSymbol(symbol);
 		int cipher = endpoint.encode(qdSymbol);
+
+		// check if symbol is accepted by filter
+		if (!filter.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();
 		if (lastEventsProcessor.ticker.getDataIfAvailable(lb.owner, delegate.getRecord(), cipher, qdSymbol))
 			return Promise.completed(delegate.createEvent(symbol, lb.owner.cursor()));
 		// not found -- need to subscribe
@@ -202,20 +241,28 @@ public <E extends LastingEvent<?>> List<Promise<E>> getLastEventsPromises(Class<
 		if (eventType == null)
 			throw new NullPointerException();
 		List<Promise<E>> result = new ArrayList<>(symbols.size());
-		assert lastEventsProcessor != null; // if we have delegate
+
 		LocalAddBatch lb = null;
 		for (Object symbol : symbols) {
 			EventDelegate<E> delegate = getLastingEventDelegateOrNull(eventType, symbol);
 			if (delegate == null) {
 				result.add(Promise.failed(new IllegalArgumentException(INVALID_EVENT_MSG)));
 				continue;
 			}
+			assert lastEventsProcessor != null; // if we have delegate
+			String qdSymbol = delegate.getQDSymbolByEventSymbol(symbol);
+			int cipher = endpoint.encode(qdSymbol);
+
+			// check if symbol is accepted by filter
+			if (!filter.accept(QDContract.TICKER, delegate.getRecord(), cipher, qdSymbol)) {
+				result.add(Promise.failed(new CancellationException("cancel")));
+				continue;
+			}
+
 			if (lb == null) {
 				lb = getLocalAddBatch();
 				lb.subscribeStartBatch();
 			}
-			String qdSymbol = delegate.getQDSymbolByEventSymbol(symbol);
-			int cipher = endpoint.encode(qdSymbol);
 			LastEventPromise<E> promise = new LastEventPromise<>(symbol, delegate, cipher, qdSymbol);
 			result.add(promise);
 			// optimization for single event -- check that it is immediately available without subscription
@@ -351,6 +398,10 @@ private <E extends IndexedEvent<?>> List<E> fetchFromHistoryIfSubscribed(Object
 		QDHistory history = (QDHistory)endpoint.getCollector(QDContract.HISTORY); // assert != null (if we have delegate)
 		String qdSymbol = delegate.getQDSymbolByEventSymbol(symbol);
 		int cipher = endpoint.encode(qdSymbol);
+
+		// check if symbol is accepted by filter
+		if (!filter.accept(QDContract.HISTORY, delegate.getRecord(), cipher, qdSymbol))
+			return Collections.emptyList();
 		// check subscription
 		if (!history.isSubscribed(delegate.getRecord(), cipher, qdSymbol, fromQDTime))
 			return null; // not subscribed
@@ -366,6 +417,10 @@ private <E extends IndexedEvent<?>> Promise<List<E>> fetchOrSubscribeFromHistory
 		QDHistory history = (QDHistory)endpoint.getCollector(QDContract.HISTORY); // assert != null (if we have delegate)
 		String qdSymbol = delegate.getQDSymbolByEventSymbol(symbol);
 		int cipher = endpoint.encode(qdSymbol);
+
+		// check if symbol is accepted by filter
+		if (!filter.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);
 		history.examineData(delegate.getRecord(), cipher, qdSymbol, fetchQDTime, toQDTime, fetch);
@@ -1074,10 +1129,11 @@ private class LastEventsProcessor extends RecordProcessor {
 		final QDTicker ticker;
 		final QDAgent tickerAgent;
 
-		LastEventsProcessor(QDTicker ticker) {
+		LastEventsProcessor(QDTicker ticker, QDFilter filter) {
 			super(endpoint.getOrCreateExecutor());
 			this.ticker = ticker;
 			tickerAgent = ticker.agentBuilder() // anonymous, so that JMX beans are not registered (see QD-445)
+				.withFilter(filter)
 				.withAttachmentStrategy(LAST_EVENT_ATTACHMENT_STRATEGY)
 				.build();
 		}
@@ -1087,7 +1143,6 @@ void start() {
 		}
 
 		void close() {
-			assert endpoint.isClosed(); // should have been already closed
 			stopProcessing();
 			tickerAgent.close();
 		}

dxfeed-impl/src/test/java/com/dxfeed/api/test/DXFeedFilterTest.java

@@ -0,0 +1,114 @@
+/*
+ * !++
+ * QDS - Quick Data Signalling Library
+ * !-
+ * Copyright (C) 2002 - 2019 Devexperts LLC
+ * !-
+ * This Source Code Form is subject to the terms of the Mozilla Public License, v. 2.0.
+ * If a copy of the MPL was not distributed with this file, You can obtain one at
+ * http://mozilla.org/MPL/2.0/.
+ * !__
+ */
+package com.dxfeed.api.test;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.*;
+
+import com.devexperts.qd.QDFilter;
+import com.devexperts.qd.kit.CompositeFilters;
+import com.devexperts.test.ThreadCleanCheck;
+import com.dxfeed.api.*;
+import com.dxfeed.api.impl.*;
+import com.dxfeed.event.market.TimeAndSale;
+import com.dxfeed.event.market.Trade;
+import com.dxfeed.promise.Promise;
+import junit.framework.TestCase;
+
+public class DXFeedFilterTest extends TestCase {
+
+	public static final String SYMBOL = "A";
+
+	private DXEndpointImpl endpoint;
+	private DXPublisher publisher;
+	private QDFilter filter = CompositeFilters.valueOf("!" + SYMBOL, DXFeedScheme.getInstance());
+
+	@Override
+	protected void setUp() throws Exception {
+		ThreadCleanCheck.before();
+		endpoint = (DXEndpointImpl)DXEndpoint.create(DXEndpoint.Role.LOCAL_HUB);
+		publisher = endpoint.getPublisher();
+	}
+
+	@Override
+	protected void tearDown() throws Exception {
+		endpoint.close();
+		ThreadCleanCheck.after();
+	}
+
+	@SuppressWarnings("deprecation")
+	public void testFilter() throws InterruptedException {
+		DXFeed feed = endpoint.getFeed();
+		DXFeed filterFeed = new DXFeedImpl(endpoint, filter);
+
+		BlockingQueue<Trade> tradesReceived = new LinkedBlockingDeque<>();
+		BlockingQueue<Trade> filterTradesReceived = new LinkedBlockingDeque<>();
+
+		DXFeedSubscription<Trade> sub = feed.createSubscription(Trade.class);
+		sub.addEventListener(tradesReceived::addAll);
+		sub.addSymbols(SYMBOL);
+
+		DXFeedSubscription<Trade> filterSub = filterFeed.createSubscription(Trade.class);
+		filterSub.addEventListener(filterTradesReceived::addAll);
+		filterSub.addSymbols(SYMBOL);
+
+		publisher.publishEvents(Collections.singletonList(new Trade(SYMBOL)));
+
+		Trade t = tradesReceived.poll(10, TimeUnit.SECONDS);
+		assertNotNull(t);
+		assertEquals(SYMBOL, t.getEventSymbol());
+
+		assertTrue(filterTradesReceived.isEmpty());
+	}
+
+	@SuppressWarnings("deprecation")
+	public void testSeparateFeedClose() throws InterruptedException {
+		DXFeed feed = endpoint.getFeed();
+		DXFeedImpl filterFeed = new DXFeedImpl(endpoint, filter);
+
+		BlockingQueue<Trade> tradesReceived = new LinkedBlockingDeque<>();
+		DXFeedSubscription<Trade> sub = feed.createSubscription(Trade.class);
+		sub.addEventListener(tradesReceived::addAll);
+		sub.addSymbols(SYMBOL);
+
+		DXFeedSubscription<Trade> filterSub = filterFeed.createSubscription(Trade.class);
+		filterSub.addSymbols(SYMBOL);
+		filterFeed.closeImpl();
+
+		publisher.publishEvents(Collections.singletonList(new Trade(SYMBOL)));
+
+		Trade t = tradesReceived.poll(10, TimeUnit.SECONDS);
+		assertNotNull(t);
+		assertEquals(SYMBOL, t.getEventSymbol());
+	}
+
+	@SuppressWarnings("deprecation")
+	public void testLastEventPromise() {
+		DXFeed filterFeed = new DXFeedImpl(endpoint, filter);
+		publisher.publishEvents(Collections.singletonList(new Trade(SYMBOL)));
+
+		long time = System.currentTimeMillis();
+		TimeAndSale tns = new TimeAndSale(SYMBOL);
+		tns.setTime(time);
+		publisher.publishEvents(Collections.singletonList(tns));
+
+		Promise<Trade> promise = filterFeed.getLastEventPromise(Trade.class, SYMBOL);
+		assertTrue(promise.isDone());
+		assertTrue(promise.hasException());
+
+		Promise<List<TimeAndSale>> timeSeriesPromise = filterFeed.getTimeSeriesPromise(
+			TimeAndSale.class, SYMBOL, time - 10_000, time + 10_000);
+		assertTrue(timeSeriesPromise.isDone());
+		assertTrue(timeSeriesPromise.hasException());
+	}
+}

dxfeed-ipf-filter/pom.xml

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