From fc3704f2c8e4ebca94e3a2e2e10d20a9074a3120 Mon Sep 17 00:00:00 2001 From: dxcity Date: Thu, 1 Oct 2020 21:03:18 +0300 Subject: [PATCH] tagging release 3.294 --- ReleaseNotes.txt | 8 + auth/pom.xml | 2 +- dxfeed-api/pom.xml | 2 +- dxfeed-bin/pom.xml | 2 +- dxfeed-codegen-verify/pom.xml | 2 +- dxfeed-codegen/pom.xml | 2 +- dxfeed-impl/pom.xml | 2 +- dxfeed-ipf-filter/pom.xml | 2 +- dxfeed-news/pom.xml | 2 +- dxfeed-ondemand/pom.xml | 2 +- dxfeed-plotter/pom.xml | 2 +- dxfeed-promise/pom.xml | 2 +- dxfeed-samples/pom.xml | 4 +- dxfeed-tools/pom.xml | 2 +- dxfeed-viewer-installer/pom.xml | 2 +- dxfeed-viewer/pom.xml | 2 +- dxfeed-webservice-impl/pom.xml | 2 +- dxfeed-webservice/pom.xml | 2 +- dxlib-benchmark/pom.xml | 2 +- dxlib-qd-sandbox/pom.xml | 2 +- dxlib/pom.xml | 2 +- .../com/devexperts/io/SerialClassContext.java | 48 +++--- .../com/devexperts/io/StringPrefixSet.java | 122 +++++-------- .../java/com/devexperts/util/TimePeriod.java | 10 +- license/pom.xml | 2 +- mars-sample/pom.xml | 2 +- mars/pom.xml | 2 +- pom.xml | 2 +- proto-sample/pom.xml | 2 +- proto-ssl/pom.xml | 2 +- proto/pom.xml | 2 +- qd-core/pom.xml | 2 +- .../com/devexperts/qd/qtp/AgentAdapter.java | 77 +++++++-- .../com/devexperts/qd/util/TimePeriod.java | 1 + qd-dataextractor/pom.xml | 2 +- qd-logger/pom.xml | 2 +- qd-nio/pom.xml | 2 +- qd-rmi/pom.xml | 2 +- qd-sample/pom.xml | 2 +- .../qtp/test/SubscriptionKeepAliveTest.java | 163 ++++++++++++++++++ qd-samplecert/pom.xml | 2 +- qd-servlet/pom.xml | 2 +- qd-stripe/pom.xml | 2 +- qd-tools/pom.xml | 2 +- .../devexperts/qd/tools/TimeSyncTracker.java | 8 +- qds-file/pom.xml | 2 +- qds-monitoring/pom.xml | 2 +- qds-tools/pom.xml | 2 +- qds/pom.xml | 2 +- rt-api-builder/pom.xml | 2 +- teamcity-version/pom.xml | 2 +- 51 files changed, 359 insertions(+), 166 deletions(-) create mode 100644 qd-sample/src/test/java/com/devexperts/qd/qtp/test/SubscriptionKeepAliveTest.java diff --git a/ReleaseNotes.txt b/ReleaseNotes.txt index 595d04b83..7aa6b4872 100644 --- a/ReleaseNotes.txt +++ b/ReleaseNotes.txt @@ -1,3 +1,11 @@ + +QDS 3.294: + +* [QD-1262] Handle MARS time synchronization tracker multicast initialization error on MacOS +* [QD-1257] Add connector option to ignore received subscription removal requests + - added 'subscriptionKeepAlive' parameter accepting '0' (default behavior) and 'inf' (ignores unsubscription) values +* [QD-1261] Unsynchronized usage of WeakHashMap in SerialClassContext + QDS 3.293: * [QD-1246] LongHashMap key_set,values,entry_set,mod_count fields should not be volatile diff --git a/auth/pom.xml b/auth/pom.xml index 278b670e9..284a11d10 100644 --- a/auth/pom.xml +++ b/auth/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 ../pom.xml 4.0.0 diff --git a/dxfeed-api/pom.xml b/dxfeed-api/pom.xml index 31e45dfa0..4cc099d49 100644 --- a/dxfeed-api/pom.xml +++ b/dxfeed-api/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 ../pom.xml 4.0.0 diff --git a/dxfeed-bin/pom.xml b/dxfeed-bin/pom.xml index a61883914..7830df198 100644 --- a/dxfeed-bin/pom.xml +++ b/dxfeed-bin/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 ../pom.xml 4.0.0 diff --git a/dxfeed-codegen-verify/pom.xml b/dxfeed-codegen-verify/pom.xml index 91f2633bc..c2787d618 100644 --- a/dxfeed-codegen-verify/pom.xml +++ b/dxfeed-codegen-verify/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 4.0.0 diff --git a/dxfeed-codegen/pom.xml b/dxfeed-codegen/pom.xml index 79bab7925..edfaf6a8a 100644 --- a/dxfeed-codegen/pom.xml +++ b/dxfeed-codegen/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 ../pom.xml 4.0.0 diff --git a/dxfeed-impl/pom.xml b/dxfeed-impl/pom.xml index 142bb6537..3dd1a81e9 100644 --- a/dxfeed-impl/pom.xml +++ b/dxfeed-impl/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 4.0.0 diff --git a/dxfeed-ipf-filter/pom.xml b/dxfeed-ipf-filter/pom.xml index 7516c319d..d03147257 100644 --- a/dxfeed-ipf-filter/pom.xml +++ b/dxfeed-ipf-filter/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 4.0.0 diff --git a/dxfeed-news/pom.xml b/dxfeed-news/pom.xml index 08b3227ec..4a3bc6c50 100644 --- a/dxfeed-news/pom.xml +++ b/dxfeed-news/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 4.0.0 diff --git a/dxfeed-ondemand/pom.xml b/dxfeed-ondemand/pom.xml index 2a0419359..1d5910eac 100644 --- a/dxfeed-ondemand/pom.xml +++ b/dxfeed-ondemand/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 4.0.0 diff --git a/dxfeed-plotter/pom.xml b/dxfeed-plotter/pom.xml index 141d379a8..7daf9d671 100644 --- a/dxfeed-plotter/pom.xml +++ b/dxfeed-plotter/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 4.0.0 diff --git a/dxfeed-promise/pom.xml b/dxfeed-promise/pom.xml index 0cb9be294..1629298d2 100644 --- a/dxfeed-promise/pom.xml +++ b/dxfeed-promise/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 4.0.0 diff --git a/dxfeed-samples/pom.xml b/dxfeed-samples/pom.xml index fe3198cf3..70d2032b3 100644 --- a/dxfeed-samples/pom.xml +++ b/dxfeed-samples/pom.xml @@ -14,13 +14,13 @@ QD com.devexperts.qd - 3.293 + 3.294 ../pom.xml 4.0.0 dxfeed-samples - 3.293 + 3.294 jar diff --git a/dxfeed-tools/pom.xml b/dxfeed-tools/pom.xml index a6d7e7afa..b6231bbaa 100644 --- a/dxfeed-tools/pom.xml +++ b/dxfeed-tools/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 ../pom.xml 4.0.0 diff --git a/dxfeed-viewer-installer/pom.xml b/dxfeed-viewer-installer/pom.xml index a882cf888..30868e169 100644 --- a/dxfeed-viewer-installer/pom.xml +++ b/dxfeed-viewer-installer/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 ../pom.xml 4.0.0 diff --git a/dxfeed-viewer/pom.xml b/dxfeed-viewer/pom.xml index 8b3004833..556b1dd7b 100644 --- a/dxfeed-viewer/pom.xml +++ b/dxfeed-viewer/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 ../pom.xml 4.0.0 diff --git a/dxfeed-webservice-impl/pom.xml b/dxfeed-webservice-impl/pom.xml index 7073c934f..0dcb29b78 100644 --- a/dxfeed-webservice-impl/pom.xml +++ b/dxfeed-webservice-impl/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 4.0.0 diff --git a/dxfeed-webservice/pom.xml b/dxfeed-webservice/pom.xml index 2832d9f9e..f9d70212b 100644 --- a/dxfeed-webservice/pom.xml +++ b/dxfeed-webservice/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 4.0.0 diff --git a/dxlib-benchmark/pom.xml b/dxlib-benchmark/pom.xml index 855ca0ede..d972715d7 100644 --- a/dxlib-benchmark/pom.xml +++ b/dxlib-benchmark/pom.xml @@ -2,7 +2,7 @@ QD com.devexperts.qd - 3.293 + 3.294 ../pom.xml 4.0.0 diff --git a/dxlib-qd-sandbox/pom.xml b/dxlib-qd-sandbox/pom.xml index 0b01b1eba..aae215113 100644 --- a/dxlib-qd-sandbox/pom.xml +++ b/dxlib-qd-sandbox/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 ../pom.xml 4.0.0 diff --git a/dxlib/pom.xml b/dxlib/pom.xml index 4e14a1a98..3c84ed984 100644 --- a/dxlib/pom.xml +++ b/dxlib/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 ../pom.xml 4.0.0 diff --git a/dxlib/src/main/java/com/devexperts/io/SerialClassContext.java b/dxlib/src/main/java/com/devexperts/io/SerialClassContext.java index 2c13f86b1..079d5d970 100644 --- a/dxlib/src/main/java/com/devexperts/io/SerialClassContext.java +++ b/dxlib/src/main/java/com/devexperts/io/SerialClassContext.java @@ -21,9 +21,11 @@ import java.net.URL; import java.nio.charset.StandardCharsets; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.List; +import java.util.Map; import java.util.Objects; import java.util.WeakHashMap; @@ -48,7 +50,8 @@ public class SerialClassContext { */ public static final String DEFAULT_BLACK_LIST_NAME = "/META-INF/SerialClassBlacklist.txt"; - private static final WeakHashMap defaultSerialContextMap = new WeakHashMap<>(); + private static final Map defaultSerialContextMap = + Collections.synchronizedMap(new WeakHashMap<>()); private static final String ARRAY_PREFIX = "["; @@ -79,45 +82,38 @@ public static SerialClassContext createSerialClassContext(ClassLoader loader, Co * @return serial class context. */ public static SerialClassContext getDefaultSerialContext(ClassLoader loader) { - if (loader == null) - loader = ClassUtil.resolveContextClassLoader(null); - SerialClassContext defaultContext = defaultSerialContextMap.get(loader); - if (defaultContext == null) { - StringPrefixSet whitelist = readPrefixSet(loader, DEFAULT_WHITE_LIST_NAME, StringPrefixSet.ANYTHING_SET); - StringPrefixSet blacklist = readPrefixSet(loader, DEFAULT_BLACK_LIST_NAME, StringPrefixSet.NOTHING_SET); - defaultContext = new SerialClassContext(loader, whitelist, blacklist); - defaultSerialContextMap.put(loader, defaultContext); - } - return defaultContext; + return defaultSerialContextMap.computeIfAbsent( + ClassUtil.resolveContextClassLoader(loader), + SerialClassContext::readSerialClassContext); + } + + private static SerialClassContext readSerialClassContext(ClassLoader cl) { + StringPrefixSet whitelist = readPrefixSet(cl, DEFAULT_WHITE_LIST_NAME, StringPrefixSet.ANYTHING_SET); + StringPrefixSet blacklist = readPrefixSet(cl, DEFAULT_BLACK_LIST_NAME, StringPrefixSet.NOTHING_SET); + return new SerialClassContext(cl, whitelist, blacklist); } private static StringPrefixSet readPrefixSet(ClassLoader cl, String prefixSetName, StringPrefixSet def) { - if (prefixSetName == null) - return def; - List urls = new ArrayList<>(); - if (cl == null) - cl = Thread.currentThread().getContextClassLoader(); + List urls; try { - urls.addAll(Collections.list(cl.getResources(prefixSetName))); + urls = Collections.list(cl.getResources(prefixSetName)); } catch (IOException e) { return def; } if (urls.isEmpty()) return def; - StringPrefixSet set = null; + List names = new ArrayList<>(); for (URL url : urls) { try (BufferedReader r = new BufferedReader(new InputStreamReader(url.openStream(), StandardCharsets.UTF_8))) { - for (String name; (name = r.readLine()) != null;) { - if (set == null) - set = StringPrefixSet.valueOf(name); - else - set = set.add(StringPrefixSet.valueOf(name)); + for (String line; (line = r.readLine()) != null;) { + // support multiple names on a single line for compatibility + names.addAll(Arrays.asList(line.split(StringPrefixSet.DEFAULT_NAMES_SEPARATOR))); } } catch (IOException e) { log.error("Cannot read " + LogUtil.hideCredentials(url), e); } } - return set == null ? def : set; + return names.isEmpty() ? def : StringPrefixSet.valueOf(names); } private final StringPrefixSet whitelist; @@ -157,7 +153,7 @@ public ClassLoader getClassLoader() { * @return {@code true}, if class name contained in whitelist and not contained in blacklist. * @throws NullPointerException if className is null. */ - public synchronized boolean accept(String className) { + public boolean accept(String className) { Objects.requireNonNull(className, "className"); if (className.startsWith(ARRAY_PREFIX)) return true; @@ -176,7 +172,7 @@ public synchronized boolean accept(String className) { * @throws ClassNotFoundException if the class not contained in whitelist or contained in blacklist. * @throws NullPointerException if className is null. */ - public synchronized void check(String className) throws ClassNotFoundException { + public void check(String className) throws ClassNotFoundException { Objects.requireNonNull(className, "className"); if (className.startsWith(ARRAY_PREFIX)) return; diff --git a/dxlib/src/main/java/com/devexperts/io/StringPrefixSet.java b/dxlib/src/main/java/com/devexperts/io/StringPrefixSet.java index d72fe7845..8ac716d24 100644 --- a/dxlib/src/main/java/com/devexperts/io/StringPrefixSet.java +++ b/dxlib/src/main/java/com/devexperts/io/StringPrefixSet.java @@ -18,9 +18,7 @@ import java.util.HashSet; import java.util.Iterator; import java.util.List; -import java.util.Set; import java.util.TreeSet; -import javax.annotation.Nonnull; /** * Class for working with set of string prefixes. @@ -36,21 +34,15 @@ class StringPrefixSet { /** * All possible prefixes. */ - public static final StringPrefixSet ANYTHING_SET = new StringPrefixSet(Collections.emptySet(), new TreeSet<>(Collections.singletonList(""))); + public static final StringPrefixSet ANYTHING_SET = new StringPrefixSet(new HashSet<>(), new TreeSet<>(Collections.singletonList(""))); /** * None of the possible prefix. */ - public static final StringPrefixSet NOTHING_SET = new StringPrefixSet(Collections.emptySet(), new TreeSet()); + public static final StringPrefixSet NOTHING_SET = new StringPrefixSet(new HashSet<>(), new TreeSet<>()); private static final String ANYTHING_SYMBOL = "*"; - @Nonnull - private final TreeSet prefixSet = new TreeSet<>(); - - @Nonnull - private final Set fullNamesSet = new HashSet<>(); - /** * Constructs set of prefix by names and separators. Names can use wildcard only symbol "*". *

Syntax of wildcard is the same as the specifications of {@code java import ...} . @@ -94,50 +86,40 @@ static StringPrefixSet valueOf(String names) { static StringPrefixSet valueOf(Collection names) { if (names == null) return NOTHING_SET; - boolean isAnything = false; - Set fullNamesSet = new HashSet<>(); - TreeSet prefixTree = new TreeSet<>(); - int i = 0; + HashSet fullNamesSet = new HashSet<>(); + TreeSet prefixSet = new TreeSet<>(); + int i = -1; for (String name : names) { + i++; if (name.length() == 0) continue; int index = name.indexOf(ANYTHING_SYMBOL); if (index == -1) { fullNamesSet.add(name); - continue; + } else { + if (index != name.length() - ANYTHING_SYMBOL.length()) + throw new IllegalArgumentException("Name at index " + i + " has wrong format: " + name); + prefixSet.add(name.substring(0, index)); } - if (index != name.length() - 1) - throw new IllegalArgumentException("Name at number " + i + " has a wrong format: " + name); - if (index == 0) { - isAnything = true; - continue; - } - prefixTree.add(name.substring(0, name.length() - 1)); - i++; } - if (isAnything) - return ANYTHING_SET; - return optimize(fullNamesSet, prefixTree); + return optimize(fullNamesSet, prefixSet); } - private static StringPrefixSet optimize(Set fullNamesSet, TreeSet prefixSet) { - Iterator it = prefixSet.iterator(); - String cur; - String last = null; - if (it.hasNext()) - last = it.next(); - while (it.hasNext()) { - cur = it.next(); - if (cur.startsWith(last)) - it.remove(); - else - last = cur; + private static StringPrefixSet optimize(HashSet fullNamesSet, TreeSet prefixSet) { + if (prefixSet.size() > 1) { + Iterator it = prefixSet.iterator(); + String last = it.next(); + while (it.hasNext()) { + String cur = it.next(); + if (cur.startsWith(last)) + it.remove(); + else + last = cur; + } } - it = fullNamesSet.iterator(); - String prefix; - while (it.hasNext()) { - cur = it.next(); - prefix = prefixSet.floor(cur); + for (Iterator it = fullNamesSet.iterator(); it.hasNext();) { + String cur = it.next(); + String prefix = prefixSet.floor(cur); if (prefix != null && cur.startsWith(prefix)) it.remove(); } @@ -150,14 +132,13 @@ private static StringPrefixSet optimize(Set fullNamesSet, TreeSet fullNamesSet, @Nonnull TreeSet prefixSet) { - this.fullNamesSet.addAll(fullNamesSet); - this.prefixSet.addAll(prefixSet); - } - private StringPrefixSet(StringPrefixSet prefixSet) { - this.fullNamesSet.addAll(prefixSet.fullNamesSet); - this.prefixSet.addAll(prefixSet.prefixSet); + private final HashSet fullNamesSet; + private final TreeSet prefixSet; + + private StringPrefixSet(HashSet fullNamesSet, TreeSet prefixSet) { + this.fullNamesSet = fullNamesSet; + this.prefixSet = prefixSet; } /** @@ -165,7 +146,6 @@ private StringPrefixSet(StringPrefixSet prefixSet) { * * @param other set of prefix * @return the union of two sets of prefix. - * @throws IllegalArgumentException if the sets are different delimiter symbol between the prefix. */ StringPrefixSet add(StringPrefixSet other) { if (this == ANYTHING_SET || other == ANYTHING_SET) @@ -179,13 +159,13 @@ StringPrefixSet add(StringPrefixSet other) { prefixSetUnion.addAll(prefixSet); return optimize(fullNamesSetUnion, prefixSetUnion); } - return this == NOTHING_SET ? new StringPrefixSet(other) : new StringPrefixSet(this); + return this == NOTHING_SET ? other : this; } StringPrefixSet copy() { if (this == ANYTHING_SET || this == NOTHING_SET) return this; - return new StringPrefixSet(this); + return new StringPrefixSet(fullNamesSet, prefixSet); } /** @@ -195,9 +175,9 @@ StringPrefixSet copy() { * @return {@code true} if name is contained in list. */ boolean accept(String name) { - if (isAnything()) + if (this == ANYTHING_SET) return true; - if (isNothing()) + if (this == NOTHING_SET) return false; if (fullNamesSet.contains(name)) return true; @@ -211,7 +191,7 @@ boolean accept(String name) { * @return {@code true} if set contains any prefix name. */ boolean isAnything() { - return fullNamesSet.isEmpty() && (prefixSet.size() == 1 && prefixSet.first().length() == 0); + return this == ANYTHING_SET; } /** @@ -220,18 +200,17 @@ boolean isAnything() { * @return {@code true} if set not contains any prefix name. */ boolean isNothing() { - return fullNamesSet.isEmpty() && prefixSet.isEmpty(); + return this == NOTHING_SET; } @Override public boolean equals(Object o) { - if (this == o) + if (o == this) return true; - if (o == null || !(o instanceof StringPrefixSet)) + if (!(o instanceof StringPrefixSet)) return false; StringPrefixSet other = (StringPrefixSet) o; - return fullNamesSet.equals(other.fullNamesSet) - && prefixSet.equals(other.prefixSet); + return fullNamesSet.equals(other.fullNamesSet) && prefixSet.equals(other.prefixSet); } @Override @@ -241,24 +220,17 @@ public int hashCode() { @Override public String toString() { - String className = "SerialClassList{"; - if (this == NOTHING_SET) - return className + "NOTHING}"; if (this == ANYTHING_SET) - return className + "ANYTHING}"; - return className + "prefixes=" + prefixSet - + ", full names = " + new TreeSet<>(fullNamesSet) - + "}"; + return "StringPrefixSet{ANYTHING}"; + if (this == NOTHING_SET) + return "StringPrefixSet{NOTHING}"; + return "StringPrefixSet{prefixes=" + prefixSet + ", full names = " + new TreeSet<>(fullNamesSet) + "}"; } public List getList() { - List result = new ArrayList<>(); - if (fullNamesSet != null) - result.addAll(fullNamesSet); - if (prefixSet != null) { - for (String prefix : prefixSet) - result.add(prefix + ANYTHING_SYMBOL); - } + List result = new ArrayList<>(fullNamesSet); + for (String prefix : prefixSet) + result.add(prefix + ANYTHING_SYMBOL); return result; } } diff --git a/dxlib/src/main/java/com/devexperts/util/TimePeriod.java b/dxlib/src/main/java/com/devexperts/util/TimePeriod.java index 34db080b0..9c6b73b97 100644 --- a/dxlib/src/main/java/com/devexperts/util/TimePeriod.java +++ b/dxlib/src/main/java/com/devexperts/util/TimePeriod.java @@ -29,6 +29,11 @@ public class TimePeriod implements Serializable { */ public static final TimePeriod UNLIMITED = new TimePeriod(Long.MAX_VALUE); + /** + * Input representation of unlimited interval + */ + private static final String UNLIMITED_STR = "inf"; + /** * Returns TimePeriod with value milliseconds. * @@ -36,7 +41,7 @@ public class TimePeriod implements Serializable { * @return TimePeriod with value milliseconds. */ public static TimePeriod valueOf(long value) { - return value == 0 ? ZERO : new TimePeriod(value); + return value == 0 ? ZERO : (value == Long.MAX_VALUE ? UNLIMITED : new TimePeriod(value)); } /** @@ -49,6 +54,7 @@ public static TimePeriod valueOf(long value) { *

  • Letter "S" can be also omitted. In this case last number will be supposed to be seconds. *
  • Number of seconds can be fractional. So it is possible to define duration accurate within milliseconds. *
  • Every part can be omitted. It is supposed that it's value is zero then. + *
  • String "inf" recognized as unlimited period. * * * @param value string representation @@ -68,6 +74,8 @@ protected TimePeriod(long value) { protected static long parse(String value) { try { + if (UNLIMITED_STR.equalsIgnoreCase(value)) + return UNLIMITED.getTime(); boolean metAnyPart = false; value = value.toUpperCase() + '#'; long res = 0; diff --git a/license/pom.xml b/license/pom.xml index e9b45b1de..82ec9d071 100644 --- a/license/pom.xml +++ b/license/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 ../pom.xml 4.0.0 diff --git a/mars-sample/pom.xml b/mars-sample/pom.xml index 91b9c6c27..855a3d006 100644 --- a/mars-sample/pom.xml +++ b/mars-sample/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 ../pom.xml 4.0.0 diff --git a/mars/pom.xml b/mars/pom.xml index c7fee6f1b..16f21e32f 100644 --- a/mars/pom.xml +++ b/mars/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 ../pom.xml 4.0.0 diff --git a/pom.xml b/pom.xml index ea150d39f..520182631 100644 --- a/pom.xml +++ b/pom.xml @@ -16,7 +16,7 @@ com.devexperts.qd QD pom - 3.293 + 3.294 2002 diff --git a/proto-sample/pom.xml b/proto-sample/pom.xml index f55d51251..3123570b2 100644 --- a/proto-sample/pom.xml +++ b/proto-sample/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 ../pom.xml 4.0.0 diff --git a/proto-ssl/pom.xml b/proto-ssl/pom.xml index 8961a2e5d..9f8c1e4ce 100644 --- a/proto-ssl/pom.xml +++ b/proto-ssl/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 ../pom.xml 4.0.0 diff --git a/proto/pom.xml b/proto/pom.xml index 49e708a5b..aeff5e672 100644 --- a/proto/pom.xml +++ b/proto/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 ../pom.xml 4.0.0 diff --git a/qd-core/pom.xml b/qd-core/pom.xml index dfc743de4..8b8f22f49 100644 --- a/qd-core/pom.xml +++ b/qd-core/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 ../pom.xml 4.0.0 diff --git a/qd-core/src/main/java/com/devexperts/qd/qtp/AgentAdapter.java b/qd-core/src/main/java/com/devexperts/qd/qtp/AgentAdapter.java index f84b59aea..1f0005fa7 100644 --- a/qd-core/src/main/java/com/devexperts/qd/qtp/AgentAdapter.java +++ b/qd-core/src/main/java/com/devexperts/qd/qtp/AgentAdapter.java @@ -26,6 +26,9 @@ import com.devexperts.qd.SubscriptionFilter; import com.devexperts.qd.SubscriptionIterator; import com.devexperts.qd.kit.CompositeFilters; +import com.devexperts.qd.ng.EventFlag; +import com.devexperts.qd.ng.RecordBuffer; +import com.devexperts.qd.ng.RecordCursor; import com.devexperts.qd.ng.RecordProvider; import com.devexperts.qd.ng.RecordSource; import com.devexperts.qd.qtp.auth.BasicChannelShaperFactory; @@ -43,6 +46,7 @@ import java.util.HashMap; import java.util.HashSet; import java.util.Map; +import java.util.Objects; import java.util.concurrent.Executor; import javax.annotation.Nonnull; @@ -86,6 +90,14 @@ public static class Factory extends MessageAdapter.AbstractFactory { */ private Executor subscriptionExecutor; + /** + * Subscription keep-alive period. + *

    If more than zero, unsubscription requests will be delayed for a specified period to amortize + * fast unsub/sub sequences. + *

    NOTE: For the moment only zero and unlimited periods are supported. + */ + private TimePeriod subscriptionKeepAlive = TimePeriod.ZERO; + public Factory(QDTicker ticker, QDStream stream, QDHistory history, SubscriptionFilter filter) { super(ticker, stream, history, filter); } @@ -165,6 +177,22 @@ public synchronized void setSubscriptionExecutor(Executor subscriptionExecutor) rebuildChannels(); } + @Nonnull + public synchronized TimePeriod getSubscriptionKeepAlive() { + return subscriptionKeepAlive; + } + + @Configurable(description = "subscription keep-alive period (0 or 'inf')") + public synchronized void setSubscriptionKeepAlive(TimePeriod keepAlive) { + Objects.requireNonNull(keepAlive); + if (subscriptionKeepAlive.equals(keepAlive)) + return; + if (!TimePeriod.ZERO.equals(keepAlive) && !TimePeriod.UNLIMITED.equals(keepAlive)) + throw new IllegalArgumentException("Only zero or infinite supported"); + this.subscriptionKeepAlive = keepAlive; + rebuildChannels(); + } + synchronized Executor getOrCreateSubscriptionExecutor() { if (subscriptionExecutor != null) return subscriptionExecutor; @@ -202,6 +230,7 @@ public MessageAdapter createAdapter(QDStats stats) { private final DataScheme scheme; private final QDFilter filter; // @NotNull private AgentAdapter.Factory factory; + private boolean skipRemoveSubscription = false; // current implementation supports only infinite keep-alive period final QDFilter[] peerFilter = new QDFilter[N_CONTRACTS]; // filters received from remote peer in DESCRIBE PROTOCOL message @@ -347,6 +376,7 @@ public void updateChannel(ChannelShaper shaper) {} private void setAgentFactory(AgentAdapter.Factory factory) { this.factory = factory; + skipRemoveSubscription = TimePeriod.UNLIMITED.equals(factory.getSubscriptionKeepAlive()); } Factory getAgentFactory() { @@ -366,27 +396,46 @@ protected void processSubscription(SubscriptionIterator iterator, MessageType me throw new IllegalArgumentException(message.toString()); if (!isAlive()) { reportIgnoredMessage("Adapter is " + getStatus(), message); - } else { - boolean hasContract = false; - QDContract contract = message.getContract(); + } else if (!skipRemoveSubscription || !message.isSubscriptionRemove()) { RecordSource sub = LegacyAdapter.of(iterator); - long initialPosition = sub.getPosition(); - for (int i = 0; i < shapers.length; i++) { - ChannelShaper shaper = shapers[i]; - if (shaper.getContract() != contract) - continue; - hasContract = true; - AgentChannel channel = getOrCreateChannelAt(i); - sub.setPosition(initialPosition); - channel.processSubscription(message, sub); + if (skipRemoveSubscription) { + RecordBuffer buf = skipRemoveSubscription(sub); + processSubscription(buf, message); + buf.release(); + } else { + processSubscription(sub, message); } LegacyAdapter.release(iterator, sub); - if (!hasContract) - reportIgnoredMessage("Contract is not supported", message); } SubscriptionConsumer.VOID.processSubscription(iterator); // silently ignore all remaining data if it was not processed } + private void processSubscription(RecordSource sub, MessageType message) { + QDContract contract = message.getContract(); + boolean hasContract = false; + long initialPosition = sub.getPosition(); + for (int i = 0; i < shapers.length; i++) { + ChannelShaper shaper = shapers[i]; + if (shaper.getContract() != contract) + continue; + hasContract = true; + AgentChannel channel = getOrCreateChannelAt(i); + sub.setPosition(initialPosition); + channel.processSubscription(message, sub); + } + if (!hasContract) + reportIgnoredMessage("Contract is not supported", message); + } + + private RecordBuffer skipRemoveSubscription(RecordSource sub) { + RecordBuffer buf = RecordBuffer.getInstance(sub.getMode()); + for (RecordCursor cur; (cur = sub.next()) != null; ) { + if (!EventFlag.REMOVE_SYMBOL.in(cur.getEventFlags())) + buf.append(cur); + } + return buf; + } + private AgentChannel getOrCreateChannelAt(int i) { AgentChannel channel = channels[i]; if (channel == null) { diff --git a/qd-core/src/main/java/com/devexperts/qd/util/TimePeriod.java b/qd-core/src/main/java/com/devexperts/qd/util/TimePeriod.java index 6789e4f12..c6a41ebb4 100644 --- a/qd-core/src/main/java/com/devexperts/qd/util/TimePeriod.java +++ b/qd-core/src/main/java/com/devexperts/qd/util/TimePeriod.java @@ -38,6 +38,7 @@ public static TimePeriod valueOf(long value) { *

  • Letter "S" can be also omitted. In this case last number will be supposed to be seconds. *
  • Number of seconds can be fractional. So it is possible to define duration accurate within milliseconds. *
  • Every part can be omitted. It is supposed that it's value is zero then. + *
  • String "inf" recognized as unlimited period. * * * @param value string representation diff --git a/qd-dataextractor/pom.xml b/qd-dataextractor/pom.xml index a4708e62e..8178baaff 100644 --- a/qd-dataextractor/pom.xml +++ b/qd-dataextractor/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 4.0.0 diff --git a/qd-logger/pom.xml b/qd-logger/pom.xml index 5a2c9d826..338e6f507 100644 --- a/qd-logger/pom.xml +++ b/qd-logger/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 ../pom.xml 4.0.0 diff --git a/qd-nio/pom.xml b/qd-nio/pom.xml index c285bfff5..141af5600 100644 --- a/qd-nio/pom.xml +++ b/qd-nio/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 ../pom.xml 4.0.0 diff --git a/qd-rmi/pom.xml b/qd-rmi/pom.xml index 8a09fdd30..949348ae0 100644 --- a/qd-rmi/pom.xml +++ b/qd-rmi/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 ../pom.xml 4.0.0 diff --git a/qd-sample/pom.xml b/qd-sample/pom.xml index 4e4c59df3..f3516c9aa 100644 --- a/qd-sample/pom.xml +++ b/qd-sample/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 ../pom.xml 4.0.0 diff --git a/qd-sample/src/test/java/com/devexperts/qd/qtp/test/SubscriptionKeepAliveTest.java b/qd-sample/src/test/java/com/devexperts/qd/qtp/test/SubscriptionKeepAliveTest.java new file mode 100644 index 000000000..c94dc2c45 --- /dev/null +++ b/qd-sample/src/test/java/com/devexperts/qd/qtp/test/SubscriptionKeepAliveTest.java @@ -0,0 +1,163 @@ +/* + * !++ + * QDS - Quick Data Signalling Library + * !- + * Copyright (C) 2002 - 2020 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.devexperts.qd.qtp.test; + +import com.devexperts.logging.Logging; +import com.devexperts.qd.DataRecord; +import com.devexperts.qd.QDAgent; +import com.devexperts.qd.QDContract; +import com.devexperts.qd.QDDistributor; +import com.devexperts.qd.ng.RecordBuffer; +import com.devexperts.qd.ng.RecordCursor; +import com.devexperts.qd.ng.RecordMode; +import com.devexperts.qd.ng.RecordProvider; +import com.devexperts.qd.qtp.AgentAdapter; +import com.devexperts.qd.qtp.DistributorAdapter; +import com.devexperts.qd.qtp.MessageConnectors; +import com.devexperts.qd.qtp.MessageType; +import com.devexperts.qd.qtp.QDEndpoint; +import com.devexperts.qd.qtp.socket.ServerSocketTestHelper; +import com.devexperts.qd.stats.QDStats; +import com.devexperts.qd.test.TestDataScheme; +import com.devexperts.util.IndexedSet; +import com.dxfeed.promise.Promise; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import java.util.Collections; +import java.util.HashSet; +import java.util.UUID; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; +import java.util.stream.Stream; + +import static com.devexperts.qd.qtp.MessageType.Flag.ADDSUB; +import static com.devexperts.qd.qtp.MessageType.Flag.REMSUB; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +public class SubscriptionKeepAliveTest { + + private static final Logging log = Logging.getLogging(SubscriptionKeepAliveTest.class); + private static final long TIMEOUT_MS = 30_000; + + private static final TestDataScheme SCHEME = new TestDataScheme(); + private static final DataRecord RECORD = SCHEME.getRecord(0); + + private QDEndpoint server; + private QDEndpoint client; + + @Before + public void setUp() { + server = createEndpoint(); + client = createEndpoint(); + } + + @After + public void tearDown() { + if (client != null) + client.close(); + if (server != null) { + server.close(); + } + } + + @Test + public void testWithoutKeepAlive() throws InterruptedException { + doTestKeepAlive(false); + } + + @Test + public void testInfiniteKeepAlive() throws InterruptedException { + doTestKeepAlive(true); + } + + protected void doTestKeepAlive(boolean keepAlive) throws InterruptedException { + String testID = UUID.randomUUID().toString(); + Promise port = ServerSocketTestHelper.createPortPromise(testID); + server.addConnectors(MessageConnectors.createMessageConnectors( + new AgentAdapter.Factory(server, null), + ":0[name=" + testID + ",bindAddr=127.0.0.1,subscriptionKeepAlive=" + (keepAlive ? "inf" : "0") + "]", + QDStats.VOID)); + server.startConnectors(); + + client.addConnectors(MessageConnectors.createMessageConnectors( + new DistributorAdapter.Factory(client, null), + "127.0.0.1:" + port.await(TIMEOUT_MS, TimeUnit.MILLISECONDS), + QDStats.VOID)); + client.startConnectors(); + + ArrayBlockingQueue subQueue = new ArrayBlockingQueue<>(10); + ArrayBlockingQueue unsubQueue = new ArrayBlockingQueue<>(10); + + QDDistributor distributor = server.getTicker().distributorBuilder().build(); + distributor.getAddedRecordProvider().setRecordListener(provider -> retrieveSub(provider, subQueue, ADDSUB)); + distributor.getRemovedRecordProvider().setRecordListener(provider -> retrieveSub(provider, unsubQueue, REMSUB)); + + QDAgent agent = client.getTicker().agentBuilder().withVoidRecordListener(true).build(); + setSubscription(agent, "A", "B", "C"); + + HashSet subSet = new HashSet<>(); + consumeSub(subQueue, 3, subSet::add); + assertEquals(IndexedSet.of("A", "B", "C"), subSet); + + setSubscription(agent, "D"); + consumeSub(subQueue, 1, subSet::add); + + if (!keepAlive) { + consumeSub(unsubQueue, 3, subSet::remove); + assertEquals(IndexedSet.of("D"), subSet); + } + + Thread.sleep(100); // give a chance to receive UNSUB + assertTrue("No unsubscriptions expected", unsubQueue.isEmpty()); + assertTrue("No subscriptions expected", subQueue.isEmpty()); + } + + protected void consumeSub(BlockingQueue queue, int num, Consumer sink) throws InterruptedException { + for (int i = 0; i < num; i++) { + String s = queue.poll(TIMEOUT_MS, TimeUnit.MILLISECONDS); + assertNotNull("Expected subscription", s); + sink.accept(s); + } + } + + protected void retrieveSub(RecordProvider provider, BlockingQueue subQueue, MessageType.Flag flag) { + RecordBuffer sub = RecordBuffer.getInstance(RecordMode.SUBSCRIPTION); + provider.retrieve(sub); + sub.rewind(); + for (RecordCursor cur; (cur = sub.next()) != null; ) { + String s = cur.getDecodedSymbol(); + log.debug("[" + flag + "]: " + s); + subQueue.add(s); + } + sub.release(); + } + + protected void setSubscription(QDAgent agent, String... symbols) { + RecordBuffer sub = RecordBuffer.getInstance(RecordMode.SUBSCRIPTION); + Stream.of(symbols).forEach(s -> sub.add(RECORD, SCHEME.getCodec().encode(s), s)); + agent.setSubscription(sub); + sub.release(); + } + + protected QDEndpoint createEndpoint() { + return QDEndpoint.newBuilder() + .withScheme(SCHEME) + .withCollectors(Collections.singletonList(QDContract.TICKER)) + .build(); + } +} diff --git a/qd-samplecert/pom.xml b/qd-samplecert/pom.xml index 132f134e8..90fa66cbb 100644 --- a/qd-samplecert/pom.xml +++ b/qd-samplecert/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 ../pom.xml 4.0.0 diff --git a/qd-servlet/pom.xml b/qd-servlet/pom.xml index 0db5dda18..fa305dc13 100644 --- a/qd-servlet/pom.xml +++ b/qd-servlet/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 ../pom.xml 4.0.0 diff --git a/qd-stripe/pom.xml b/qd-stripe/pom.xml index 94253e7cf..e5f715981 100644 --- a/qd-stripe/pom.xml +++ b/qd-stripe/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 ../pom.xml 4.0.0 diff --git a/qd-tools/pom.xml b/qd-tools/pom.xml index 8993b7a0f..2a55ba8c9 100644 --- a/qd-tools/pom.xml +++ b/qd-tools/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 ../pom.xml 4.0.0 diff --git a/qd-tools/src/main/java/com/devexperts/qd/tools/TimeSyncTracker.java b/qd-tools/src/main/java/com/devexperts/qd/tools/TimeSyncTracker.java index ddc2f4ea9..df5c8bc6c 100644 --- a/qd-tools/src/main/java/com/devexperts/qd/tools/TimeSyncTracker.java +++ b/qd-tools/src/main/java/com/devexperts/qd/tools/TimeSyncTracker.java @@ -127,13 +127,9 @@ public synchronized void start() { try { socket.joinGroup(InetAddress.getByName(addr)); } catch (IOException e) { - // Workaround for JDK-8178161 (see QD-1131) - String javaVersion = System.getProperty("java.version"); + // Workaround for JDK-8178161 (see QD-1131, QD-1262) String osName = System.getProperty("os.name"); - if (e.getMessage().contains("assign requested address") && - osName.toLowerCase().startsWith("mac") && - (javaVersion.startsWith("1.8") || javaVersion.startsWith("9."))) - { + if (e.getMessage().contains("assign requested address") && osName.toLowerCase().startsWith("mac")) { QDLog.log.info("Time synchronization tracker initialization failed - unsupported on MacOS"); } else { QDLog.log.error("Failed to join multicast group for time synchronization tracker", e); diff --git a/qds-file/pom.xml b/qds-file/pom.xml index fbeacbb70..253075c08 100644 --- a/qds-file/pom.xml +++ b/qds-file/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 ../pom.xml 4.0.0 diff --git a/qds-monitoring/pom.xml b/qds-monitoring/pom.xml index ce50015cb..e7388162b 100644 --- a/qds-monitoring/pom.xml +++ b/qds-monitoring/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 ../pom.xml 4.0.0 diff --git a/qds-tools/pom.xml b/qds-tools/pom.xml index 391328745..32fce7a5d 100644 --- a/qds-tools/pom.xml +++ b/qds-tools/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 ../pom.xml 4.0.0 diff --git a/qds/pom.xml b/qds/pom.xml index bebb55d7f..54c4378e7 100644 --- a/qds/pom.xml +++ b/qds/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 ../pom.xml 4.0.0 diff --git a/rt-api-builder/pom.xml b/rt-api-builder/pom.xml index ca8f84278..052c795ef 100644 --- a/rt-api-builder/pom.xml +++ b/rt-api-builder/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 4.0.0 diff --git a/teamcity-version/pom.xml b/teamcity-version/pom.xml index cbddda2ba..3fcd8d7a3 100644 --- a/teamcity-version/pom.xml +++ b/teamcity-version/pom.xml @@ -14,7 +14,7 @@ QD com.devexperts.qd - 3.293 + 3.294 ../pom.xml 4.0.0