You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@edgent.apache.org by dl...@apache.org on 2016/05/26 12:33:10 UTC
[3/9] incubator-quarks git commit: relocate StreamScope to utils /
quarks.streamscope
relocate StreamScope to utils / quarks.streamscope
Project: http://git-wip-us.apache.org/repos/asf/incubator-quarks/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quarks/commit/4b36d473
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quarks/tree/4b36d473
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quarks/diff/4b36d473
Branch: refs/heads/master
Commit: 4b36d47374a1b36d53cbbd574bfceea5c1ebcaba
Parents: e10447c
Author: Dale LaBossiere <dl...@us.ibm.com>
Authored: Thu May 12 09:06:51 2016 -0400
Committer: Dale LaBossiere <dl...@us.ibm.com>
Committed: Wed May 18 16:03:04 2016 -0400
----------------------------------------------------------------------
.../java/quarks/oplet/plumbing/StreamScope.java | 44 --
.../quarks/topology/plumbing/StreamScope.java | 415 -------------------
.../topology/plumbing/StreamScopeRegistry.java | 145 -------
.../java/quarks/test/topology/PlumbingTest.java | 289 -------------
build.xml | 4 +-
providers/development/build.xml | 2 +
.../development/DevelopmentProvider.java | 8 +-
.../streamscope/DevelopmentStreamScopeTest.java | 25 ++
utils/.classpath | 2 +
utils/streamscope/build.xml | 44 ++
.../java/quarks/streamscope/StreamScope.java | 415 +++++++++++++++++++
.../quarks/streamscope/StreamScopeRegistry.java | 145 +++++++
.../quarks/streamscope/oplets/StreamScope.java | 44 ++
.../test/streamscope/StreamScopeTest.java | 325 +++++++++++++++
14 files changed, 1009 insertions(+), 898 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/4b36d473/api/oplet/src/main/java/quarks/oplet/plumbing/StreamScope.java
----------------------------------------------------------------------
diff --git a/api/oplet/src/main/java/quarks/oplet/plumbing/StreamScope.java b/api/oplet/src/main/java/quarks/oplet/plumbing/StreamScope.java
deleted file mode 100644
index af97a10..0000000
--- a/api/oplet/src/main/java/quarks/oplet/plumbing/StreamScope.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
-Licensed to the Apache Software Foundation (ASF) under one
-or more contributor license agreements. See the NOTICE file
-distributed with this work for additional information
-regarding copyright ownership. The ASF licenses this file
-to you 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 quarks.oplet.plumbing;
-
-import quarks.function.Consumer;
-import quarks.oplet.functional.Peek;
-
-/**
- * A Stream "oscilloscope" oplet.
- * <P>
- * TODO remove this? Just added to help the Console specialize its presentation
- * and/or so user can differentiate a StreamScope from any other random Peek oplet use.
- * </P>
- *
- * @param <T> Type of the tuple.
- */
-public class StreamScope<T> extends Peek<T> {
- private static final long serialVersionUID = 1L;
-
- /**
- * Create a new instance.
- * @param streamScope the consumer function
- */
- public StreamScope(Consumer<T> streamScope) {
- super(streamScope);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/4b36d473/api/topology/src/main/java/quarks/topology/plumbing/StreamScope.java
----------------------------------------------------------------------
diff --git a/api/topology/src/main/java/quarks/topology/plumbing/StreamScope.java b/api/topology/src/main/java/quarks/topology/plumbing/StreamScope.java
deleted file mode 100644
index 2252e2a..0000000
--- a/api/topology/src/main/java/quarks/topology/plumbing/StreamScope.java
+++ /dev/null
@@ -1,415 +0,0 @@
-/*
-Licensed to the Apache Software Foundation (ASF) under one
-or more contributor license agreements. See the NOTICE file
-distributed with this work for additional information
-regarding copyright ownership. The ASF licenses this file
-to you 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 quarks.topology.plumbing;
-
-import java.util.Collections;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Objects;
-import java.util.concurrent.TimeUnit;
-
-import quarks.function.Consumer;
-import quarks.function.Functions;
-import quarks.function.Predicate;
-import quarks.topology.TStream;
-
-/**
- * A Stream "oscilloscope" for capturing stream tuples for analysis / debug.
- * <P>
- * A {@code StreamScope} is expected to be used as parameter to
- * {@link TStream#peek(Consumer)}.
- * </P><P>
- * A {@link TriggerManager} controls which tuples are captured.
- * A {@link BufferManager} controls the retention policy for captured tuples.
- * </P>
- * <P>
- * StreamScope instances are typically registered in and located via
- * a {@link StreamScopeRegistry}.
- * </P>
- * @see StreamScopeRegistry
- * @see quarks.providers.development.DevelopmentProvider DevelopmentProvider
- */
-public class StreamScope<T> implements Consumer<T> {
- private static final long serialVersionUID = 1L;
- private final BufferManager<T> buffer = new BufferManager<>();
- private final TriggerManager<T> trigger = new TriggerManager<>();
- private boolean isEnabled;
-
- /**
- * A captured tuple.
- * <P>
- * The Sample captures the tuple, and the system time and nanoTime
- * that the tuple was received.
- * </P>
- *
- * @param <T> Tuple type.
- */
- public static class Sample<T> {
- private final long ts;
- private final long nanoTime;
- private final T tuple;
-
- Sample(T tuple) {
- this.ts = System.currentTimeMillis();
- this.nanoTime = System.nanoTime();
- this.tuple = tuple;
- }
-
- /**
- * Capture time in msec since the epoch.
- * @return the timestamp
- * @see System#currentTimeMillis()
- */
- public long timestamp() {
- return ts;
- }
-
- /**
- * Capture time in nanoTime.
- * @return the nanoTime
- * @see System#nanoTime()
- */
- public long nanoTime() {
- return nanoTime;
- }
-
- /**
- * The captured tuple.
- * @return the tuple
- */
- public T tuple() {
- return tuple;
- }
-
- @Override
- public String toString() {
- return "ts="+ts+" nano="+nanoTime+" tuple="+tuple;
- }
- }
-
- /**
- * Control the retention of captured tuples.
- * <P>
- * Captured tuples are retained until either:
- * <ul>
- * <li>a maximum retention count is exceeded</li>
- * <li>TODO a maximum retention time is exceeded</li>
- * </ul>
- * </P><P>
- * The default configuration is a maxCount of 10.
- * </P>
- */
- public static class BufferManager<T> {
- private List<Sample<T>> buffer = Collections.emptyList();
- private int maxCount = 10;
- private long period;
- private TimeUnit unit;
-
- // TODO timer based eviction
- // TODO look at using quarks Windows / WindowImpl for the buffer.
- // Can window type/"size" be changed?
- // Does it support simultaneous maxCount & maxAge?
-
- List<Sample<T>> getSamples() {
- return Collections.unmodifiableList(buffer);
- }
-
- /**
- * Set the maximum number of tuples to retain.
- * <P>
- * The capture buffer is cleared.
- * </P>
- * @param maxCount the maximum number of tuples to retain.
- * Specify 0 to disable count based retention.
- */
- public void setMaxRetentionCount(int maxCount) {
- this.maxCount = maxCount;
- allocate();
- }
-
- /**
- * Set the maximum retention time of a tuple.
- * <P>
- * The capture buffer is cleared.
- * </P>
- * @param age the amount of time to retain a tuple.
- * Specify 0 to disable age based retention.
- * @param unit {@link TimeUnit}
- */
- public void setMaxRetentionTime(long age, TimeUnit unit) {
- if (age < 0)
- throw new IllegalArgumentException("age");
- Objects.requireNonNull(unit, "unit");
- this.period = age;
- this.unit = unit;
-
- throw new IllegalStateException("setMaxRetentionTime is NYI");
- }
-
- /**
- * Get the number of tuples in the capture buffer.
- * @return the count.
- */
- int getCount() {
- return buffer.size();
- }
-
- void release() {
- buffer = Collections.emptyList();
- }
-
- void allocate() {
- buffer = new LinkedList<>();
- }
-
- void add(Sample<T> sample) {
- if (maxCount > 0 && buffer.size() >= maxCount)
- buffer.remove(0);
- buffer.add(sample);
- }
-
- @Override
- public String toString() {
- return "size="+getCount()+" maxCount="+maxCount+" maxAge="+period+(unit==null ? "" : unit);
- }
-
- }
-
- /**
- * Control what triggers capturing of tuples.
- * <P>
- * The following modes are supported:
- * <ul>
- * <li>continuous - capture every tuple</li>
- * <li>by-count - capture every nth tuple</li>
- * <li>by-time - capture based on time elapsed since the last capture</li>
- * <li>by-Predicate - capture based on evaluating a predicate</li>
- * </ul>
- * </P><P>
- * Tuple capture can be temporarily paused via {@link TriggerManager#setPaused(boolean) setPaused}.
- * Pausing capture does not clear the capture buffer.
- * </P><P>
- * Capture processing can be automatically paused when a particular event
- * has occurred via {@link TriggerManager#setPauseOn(Predicate) setPauseOn}.
- * The pause predicate is evaluated after processing each received tuple.
- * Use {@link TriggerManager#setPaused(boolean)} setPaused} to re-enable capture
- * following a triggered pause.
- * </P><P>
- * The default configuration is {@code continuous} and not paused.
- * </P>
- */
- public static class TriggerManager<T> {
- private Predicate<T> predicate = Functions.alwaysTrue();
- private Predicate<T> pauseOnPredicate = Functions.alwaysFalse();
- private boolean paused = false;
-
- /**
- * Test if the tuple should be captured.
- * @param tuple the tuple
- * @return true to capture the tuple.
- */
- boolean test(T tuple) {
- if (paused)
- return false;
- boolean b = predicate.test(tuple);
- paused = pauseOnPredicate.test(tuple);
- return b;
- }
-
- /**
- * Set capture paused control
- * @param paused true to pause, false to clear pause.
- */
- public void setPaused(boolean paused) {
- this.paused = paused;
- }
-
- /**
- * Is capture paused?
- * @return true if paused
- */
- public boolean isPaused() {
- return paused;
- }
-
- /**
- * Set a pause-on predicate. Capture is paused if the predicate
- * returns true;
- * @param predicate the predicate
- * @see Functions#alwaysFalse()
- */
- public void setPauseOn(Predicate<T> predicate) {
- Objects.requireNonNull(predicate, "predicate");
- pauseOnPredicate = predicate;
- }
-
- /**
- * Capture every tuple.
- */
- public void setContinuous() {
- setByPredicate(Functions.alwaysTrue());
- }
-
- /**
- * Capture the first and every nth tuple
- * @param count
- */
- public void setByCount(int count) {
- setByPredicate(newByCountPredicate(count));
- }
-
- /**
- * Capture the 1st tuple and then the next tuple after {@code period}
- * {@code unit} time has elapsed since the previous captured tuple.
- *
- * @param elapsed time to delay until next capture
- * @param unit {@link TimeUnit}
- */
- public void setByTime(long elapsed, TimeUnit unit) {
- setByPredicate(newByTimePredicate(elapsed, unit));
- }
-
- /**
- * Capture a tuple if the {@code predicate} test of the tuple returns true.
- * @param predicate
- */
- public void setByPredicate(Predicate<T> predicate) {
- Objects.requireNonNull(predicate, "predicate");
- this.predicate = predicate;
- }
-
- private static <T> Predicate<T> newByCountPredicate(int count) {
- if (count < 1)
- throw new IllegalArgumentException("count");
- return new Predicate<T>() {
- private static final long serialVersionUID = 1L;
- int byCount = count;
- int curCount = -1; // capture 1st and every byCount-nth
-
- @Override
- public boolean test(T value) {
- return ++curCount % byCount == 0;
- }
- };
- }
-
- private static <T> Predicate<T> newByTimePredicate(long elapsed, TimeUnit unit) {
- if (elapsed < 1)
- throw new IllegalArgumentException("elapsed");
- Objects.requireNonNull(unit, "unit");
- return new Predicate<T>() {
- private static final long serialVersionUID = 1L;
- private long nextTime;
-
- @Override
- public boolean test(T value) {
- long now = System.currentTimeMillis();
- if (now > nextTime) {
- nextTime = now + unit.toMillis(elapsed);
- return true;
- }
- return false;
- }
- };
- }
-
- }
-
- /**
- * Create a new instance.
- * <P>
- * Sample capture is disabled.
- * </P>
- */
- public StreamScope() {
- }
-
- /**
- * Is tuple capture enabled?
- * @return true if enabled.
- */
- public boolean isEnabled() {
- return isEnabled;
- }
-
- /**
- * Enable or disable tuple capture.
- * <P>
- * Disabling releases the capture buffer.
- * </P>
- * @param isEnabled true to enable, false to disable.
- */
- public synchronized void setEnabled(boolean isEnabled) {
- if (!isEnabled)
- buffer.release();
- buffer.allocate();
- this.isEnabled = isEnabled;
- }
-
- /**
- * Get the {@link BufferManager}
- * @return the manager
- */
- public BufferManager<T> bufferMgr() {
- return buffer;
- }
-
- /**
- * Get the {@link TriggerManager}
- * @return the manager
- */
- public TriggerManager<T> triggerMgr() {
- return trigger;
- }
-
- /**
- * Get all captured tuples.
- * <P>
- * The returned samples are removed from the capture buffer.
- * </P>
- * @return unmodifiable list of captured samples
- */
- public synchronized List<Sample<T>> getSamples() {
- List<Sample<T>> tmp = buffer.getSamples();
- buffer.allocate();
- return tmp;
- }
-
- /**
- * Get the number of Samples currently captured
- * @return the count
- */
- public synchronized int getSampleCount() {
- return buffer.getCount();
- }
-
- @Override
- public synchronized void accept(T tuple) {
- if (!isEnabled())
- return;
- if (trigger.test(tuple))
- buffer.add(new Sample<T>(tuple));
- }
-
- @Override
- public String toString() {
- return "isEnabled="+isEnabled+" bufferMgr="+bufferMgr();
- }
-}
-
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/4b36d473/api/topology/src/main/java/quarks/topology/plumbing/StreamScopeRegistry.java
----------------------------------------------------------------------
diff --git a/api/topology/src/main/java/quarks/topology/plumbing/StreamScopeRegistry.java b/api/topology/src/main/java/quarks/topology/plumbing/StreamScopeRegistry.java
deleted file mode 100644
index dbfdd80..0000000
--- a/api/topology/src/main/java/quarks/topology/plumbing/StreamScopeRegistry.java
+++ /dev/null
@@ -1,145 +0,0 @@
-/*
-Licensed to the Apache Software Foundation (ASF) under one
-or more contributor license agreements. See the NOTICE file
-distributed with this work for additional information
-regarding copyright ownership. The ASF licenses this file
-to you 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 quarks.topology.plumbing;
-
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Set;
-
-/**
- * A registry for Stream "oscilloscope" {@link StreamScope} instances.
- * <P>
- * The registry contains a collection of StreamScope instances
- * that are registered by one or more names.
- * </P><P>
- * The names are: by a TStream {@link quarks.topology.TStream#alias(String) alias} or
- * by a stream's (output port's) unique identifier.
- * Static methods are provided for composing these names and extracting
- * the alias/identifier from generated names.
- * </P>
- * @see quarks.providers.development.DevelopmentProvider DevelopmentProvider
- */
-public class StreamScopeRegistry {
- private final Map<String, StreamScope<?>> byNameMap = new HashMap<>();
- private final Map<StreamScope<?>, List<String>> byStreamScopeMap = new HashMap<>();
-
- public StreamScopeRegistry() {
-
- }
-
- /** create a registration name for a stream alias */
- public static String nameByStreamAlias(String alias) {
- Objects.requireNonNull(alias, "alias");
- return "alias."+alias;
- }
-
- /** create a registration name for a stream id */
- public static String nameByStreamId(String id) {
- Objects.requireNonNull(id, "id");
- return "id."+id;
- }
-
- /** returns null if {@code name} is not a from nameByStreamAlias() */
- public static String streamAliasFromName(String name) {
- Objects.requireNonNull(name, "name");
- if (!name.startsWith("alias."))
- return null;
- return name.substring("alias.".length());
- }
-
- /** returns null if {@code name} is not a from nameByStreamId() */
- public static String streamIdFromName(String name) {
- Objects.requireNonNull(name, "name");
- if (!name.startsWith("id."))
- return null;
- return name.substring("id.".length());
- }
-
- /** A single StreamScope can be registered with multiple names.
- * @throws IllegalStateException if a registration already exists for {@code name}
- */
- public synchronized void register(String name, StreamScope<?> streamScope) {
- if (byNameMap.containsKey(name))
- throw new IllegalStateException("StreamScope already registered by name "+name);
- byNameMap.put(name, streamScope);
- List<String> names = byStreamScopeMap.get(streamScope);
- if (names == null) {
- names = new ArrayList<>(2);
- byStreamScopeMap.put(streamScope, names);
- }
- names.add(name);
- }
-
- /**
- * @param name
- * @return the StreamScope. null if name is not registered.
- */
- public synchronized StreamScope<?> lookup(String name) {
- return byNameMap.get(name);
- }
-
- /**
- * Get the registered names.
- * @return unmodifiable collection of the name.
- * The set is backed by the registry so the set may change.
- */
- public synchronized Set<String> getNames() {
- return Collections.unmodifiableSet(byNameMap.keySet());
- }
-
- /** Get registered StreamScopes and the name(s) each is registered with.
- * The map is backed by the registry so its contents may change.
- */
- public synchronized Map<StreamScope<?>, List<String>> getStreamScopes() {
- return Collections.unmodifiableMap(byStreamScopeMap);
- }
-
- /** remove the specific name registration. Other registration of the same StreamScope may still exist.
- * no-op if name is not registered.
- * @see #unregister(StreamScope)
- */
- public synchronized void unregister(String name) {
- StreamScope<?> streamScope = byNameMap.get(name);
- if (streamScope == null)
- return;
- byNameMap.remove(name);
- List<String> names = byStreamScopeMap.get(streamScope);
- names.remove(name);
- if (names.isEmpty())
- byStreamScopeMap.remove(streamScope);
- }
-
- /** remove all name registrations of the StreamScope.
- * no-op if no registrations for the StreamScope
- */
- public synchronized void unregister(StreamScope<?> streamScope) {
- List<String> names = byStreamScopeMap.get(streamScope);
- if (names == null)
- return;
- names = new ArrayList<>(names);
- for (String name : names)
- unregister(name);
- }
-
-}
-
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/4b36d473/api/topology/src/test/java/quarks/test/topology/PlumbingTest.java
----------------------------------------------------------------------
diff --git a/api/topology/src/test/java/quarks/test/topology/PlumbingTest.java b/api/topology/src/test/java/quarks/test/topology/PlumbingTest.java
index 0dfaa6c..685bf32 100644
--- a/api/topology/src/test/java/quarks/test/topology/PlumbingTest.java
+++ b/api/topology/src/test/java/quarks/test/topology/PlumbingTest.java
@@ -20,9 +20,6 @@ package quarks.test.topology;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.junit.Assume.assumeTrue;
@@ -47,9 +44,6 @@ import quarks.function.ToIntFunction;
import quarks.topology.TStream;
import quarks.topology.Topology;
import quarks.topology.plumbing.PlumbingStreams;
-import quarks.topology.plumbing.StreamScope;
-import quarks.topology.plumbing.StreamScope.Sample;
-import quarks.topology.plumbing.StreamScopeRegistry;
import quarks.topology.plumbing.Valve;
import quarks.topology.tester.Condition;
@@ -749,288 +743,5 @@ public abstract class PlumbingTest extends TopologyAbstractTest {
assertTrue("valid:" + contents.getResult(), contents.valid());
assertTrue("valid:" + resultAvailablePermits, resultAvailablePermits.equals(arrayResult));
}
-
- @Test
- public void testStreamScopeFn() throws Exception {
-
- StreamScope<Integer> ss = new StreamScope<>();
-
- List<Sample<Integer>> samples;
- Sample<Integer> sample;
-
- assertFalse(ss.isEnabled());
- assertNotNull(ss.bufferMgr());
- assertNotNull(ss.triggerMgr());
- assertEquals(0, ss.getSampleCount());
- samples = ss.getSamples();
- assertNotNull(samples);
- assertEquals(0, samples.size());
-
- // ---------------- no capture when not enabled
- ss.accept(1);
- assertEquals(0, ss.getSampleCount());
- assertEquals(0, ss.getSamples().size());
-
- ss.setEnabled(true);
-
- // ---------------- single capture
- // note: getSamples() removes captured tuples
- ss.accept(100);
- assertEquals(1, ss.getSampleCount());
- samples = ss.getSamples();
- assertEquals(1, samples.size());
- sample = samples.get(0);
- assertEquals(100, sample.tuple().intValue());
- assertTrue(sample.timestamp() != 0);
- assertTrue(sample.nanoTime() != 0);
- assertEquals(0, ss.getSampleCount());
- assertEquals(0, ss.getSamples().size());
-
- // ---------------- next capture/get; different lists
- List<Sample<Integer>> savedSamples = samples;
- ss.accept(101);
- assertEquals(1, ss.getSampleCount());
- samples = ss.getSamples();
- assertEquals(101, samples.get(0).tuple().intValue());
- assertTrue(samples != savedSamples);
- assertEquals(0, ss.getSampleCount());
- assertEquals(0, ss.getSamples().size());
-
- // ---------------- multi capture
- ss.accept(100);
- ss.accept(101);
- ss.accept(102);
- assertEquals(3, ss.getSampleCount());
- samples = ss.getSamples();
- assertEquals(100, samples.get(0).tuple().intValue());
- assertEquals(101, samples.get(1).tuple().intValue());
- assertEquals(102, samples.get(2).tuple().intValue());
- assertEquals(0, ss.getSampleCount());
- assertEquals(0, ss.getSamples().size());
-
- // ---------------- disable => clears capture buffer
- ss.accept(100);
- ss.accept(101);
- ss.accept(102);
- assertEquals(3, ss.getSampleCount());
- ss.setEnabled(false);
- assertEquals(0, ss.getSampleCount());
- assertEquals(0, ss.getSamples().size());
-
- ss.setEnabled(true);
-
- // ---------------- buffer control at the limit (no eviction)
- ss.bufferMgr().setMaxRetentionCount(3);
- ss.accept(100);
- ss.accept(101);
- ss.accept(102);
- assertEquals(3, ss.getSampleCount());
- samples = ss.getSamples();
- assertEquals(100, samples.get(0).tuple().intValue());
- assertEquals(101, samples.get(1).tuple().intValue());
- assertEquals(102, samples.get(2).tuple().intValue());
- assertEquals(0, ss.getSampleCount());
- assertEquals(0, ss.getSamples().size());
-
- // ---------------- buffer control with eviction
- ss.bufferMgr().setMaxRetentionCount(2);
- ss.accept(100);
- ss.accept(101);
- ss.accept(102);
- assertEquals(2, ss.getSampleCount());
- samples = ss.getSamples();
- assertEquals(101, samples.get(0).tuple().intValue());
- assertEquals(102, samples.get(1).tuple().intValue());
- assertEquals(0, ss.getSampleCount());
- assertEquals(0, ss.getSamples().size());
- ss.bufferMgr().setMaxRetentionCount(10);
-
- // ---------------- trigger byCount
- ss.triggerMgr().setByCount(3);
- ss.accept(100);
- ss.accept(101);
- ss.accept(102);
- ss.accept(103);
- ss.accept(104);
- ss.accept(105);
- ss.accept(106);
- assertEquals(3, ss.getSampleCount());
- samples = ss.getSamples();
- assertEquals(100, samples.get(0).tuple().intValue());
- assertEquals(103, samples.get(1).tuple().intValue());
- assertEquals(106, samples.get(2).tuple().intValue());
-
- // ---------------- trigger byPredicate
- ss.triggerMgr().setByPredicate(t -> t % 2 == 0);
- ss.accept(100);
- ss.accept(101);
- ss.accept(102);
- ss.accept(103);
- ss.accept(104);
- assertEquals(3, ss.getSampleCount());
- samples = ss.getSamples();
- assertEquals(100, samples.get(0).tuple().intValue());
- assertEquals(102, samples.get(1).tuple().intValue());
- assertEquals(104, samples.get(2).tuple().intValue());
-
- // ---------------- trigger byTime
- ss.triggerMgr().setByTime(100, TimeUnit.MILLISECONDS);
- ss.accept(100);
- ss.accept(101);
- ss.accept(102);
- Thread.sleep(110);
- ss.accept(103);
- ss.accept(104);
- ss.accept(105);
- Thread.sleep(110);
- ss.accept(106);
- assertEquals(3, ss.getSampleCount());
- samples = ss.getSamples();
- assertEquals(100, samples.get(0).tuple().intValue());
- assertEquals(103, samples.get(1).tuple().intValue());
- assertEquals(106, samples.get(2).tuple().intValue());
-
- // ---------------- trigger continuous
- ss.triggerMgr().setContinuous();
- ss.accept(100);
- ss.accept(101);
- ss.accept(102);
- assertEquals(3, ss.getSampleCount());
- samples = ss.getSamples();
- assertEquals(100, samples.get(0).tuple().intValue());
- assertEquals(101, samples.get(1).tuple().intValue());
- assertEquals(102, samples.get(2).tuple().intValue());
-
- // ---------------- trigger pause
- ss.accept(100);
- ss.accept(101);
- ss.triggerMgr().setPaused(true);
- assertTrue(ss.triggerMgr().isPaused());
- ss.accept(102);
- ss.accept(103);
- ss.triggerMgr().setPaused(false);
- assertFalse(ss.triggerMgr().isPaused());
- ss.accept(104);
- assertEquals(3, ss.getSampleCount());
- samples = ss.getSamples();
- assertEquals(100, samples.get(0).tuple().intValue());
- assertEquals(101, samples.get(1).tuple().intValue());
- assertEquals(104, samples.get(2).tuple().intValue());
-
- // ---------------- trigger pauseOn
-
- ss.triggerMgr().setPauseOn(t -> t == 102);
- ss.accept(100);
- ss.accept(101);
- ss.accept(102);
- ss.accept(103);
- ss.accept(104);
- ss.accept(105);
- assertTrue(ss.triggerMgr().isPaused());
- assertEquals(3, ss.getSampleCount());
- samples = ss.getSamples();
- assertEquals(100, samples.get(0).tuple().intValue());
- assertEquals(101, samples.get(1).tuple().intValue());
- assertEquals(102, samples.get(2).tuple().intValue());
- ss.triggerMgr().setPaused(false);
- ss.accept(1000);
- ss.accept(1010);
- ss.accept(102);
- ss.accept(1030);
- assertEquals(3, ss.getSampleCount());
- samples = ss.getSamples();
- assertEquals(1000, samples.get(0).tuple().intValue());
- assertEquals(1010, samples.get(1).tuple().intValue());
- assertEquals(102, samples.get(2).tuple().intValue());
-
- }
-
- @Test
- public void testStreamScopeRegistry() throws Exception {
-
- StreamScope<Integer> ss1 = new StreamScope<>();
- StreamScope<Integer> ss2 = new StreamScope<>();
-
- StreamScopeRegistry rgy = new StreamScopeRegistry();
-
- assertNotNull(rgy.getNames());
- assertEquals(0, rgy.getNames().size());
- assertNotNull(rgy.getStreamScopes());
- assertEquals(0, rgy.getStreamScopes().size());
- assertNull(rgy.lookup("xyzzy"));
- rgy.unregister("xyzzy");
- rgy.unregister(ss1);
-
- // ---------- name generation / parse functions
- String alias1Name = StreamScopeRegistry.nameByStreamAlias("alias1");
- assertNotNull(alias1Name);
- String alias2Name = StreamScopeRegistry.nameByStreamAlias("alias2");
- assertNotNull(alias2Name);
- assertFalse(alias1Name.equals(alias2Name));
- String alias1 = StreamScopeRegistry.streamAliasFromName(alias1Name);
- assertEquals("alias1", alias1);
-
- String id1Name = StreamScopeRegistry.nameByStreamId("id1");
- assertNotNull(id1Name);
- String id2Name = StreamScopeRegistry.nameByStreamId("id2");
- assertNotNull(id2Name);
- assertFalse(id1Name.equals(id2Name));
- String id1 = StreamScopeRegistry.streamIdFromName(id1Name);
- assertEquals("id1", id1);
-
- assertFalse(StreamScopeRegistry.nameByStreamAlias("1")
- .equals(StreamScopeRegistry.nameByStreamId("1")));
-
- // ---------- register
- rgy.register(alias1Name, ss1);
- rgy.register(alias2Name, ss2);
- rgy.register(id2Name, ss2);
-
- // ---------- lookup
- assertSame(ss1, rgy.lookup(alias1Name));
- assertSame(ss2, rgy.lookup(alias2Name));
- assertSame(ss2, rgy.lookup(id2Name));
-
- // ----------- getNames
- assertEquals(3, rgy.getNames().size());
- assertTrue(rgy.getNames().contains(alias1Name));
- assertFalse(rgy.getNames().contains(id1Name));
- assertTrue(rgy.getNames().contains(alias2Name));
- assertTrue(rgy.getNames().contains(id2Name));
-
- // ----------- getStreamScopes
- assertEquals(2, rgy.getStreamScopes().keySet().size());
- assertTrue(rgy.getStreamScopes().keySet().contains(ss1));
- assertTrue(rgy.getStreamScopes().keySet().contains(ss2));
- assertEquals(1, rgy.getStreamScopes().get(ss1).size());
- assertTrue(rgy.getStreamScopes().get(ss1).contains(alias1Name));
- assertEquals(2, rgy.getStreamScopes().get(ss2).size());
- assertTrue(rgy.getStreamScopes().get(ss2).contains(alias2Name));
- assertTrue(rgy.getStreamScopes().get(ss2).contains(id2Name));
-
- // ---------- unregister
- rgy.unregister(alias1Name);
- assertNull(rgy.lookup(alias1Name));
- assertEquals(2, rgy.getNames().size());
- assertFalse(rgy.getNames().contains(alias1Name));
- assertFalse(rgy.getStreamScopes().keySet().contains(ss1));
- assertTrue(rgy.getStreamScopes().keySet().contains(ss2));
-
- rgy.unregister(alias2Name);
- assertNull(rgy.lookup(alias2Name));
- assertEquals(1, rgy.getNames().size());
- assertFalse(rgy.getNames().contains(alias2Name));
- assertTrue(rgy.getStreamScopes().keySet().contains(ss2));
- assertSame(ss2, rgy.lookup(id2Name));
- rgy.unregister(id2Name);
- assertEquals(0, rgy.getNames().size());
- assertEquals(0, rgy.getStreamScopes().keySet().size());
-
- rgy.register(alias2Name, ss2);
- rgy.register(id2Name, ss2);
- rgy.unregister(ss2);
- assertEquals(0, rgy.getNames().size());
- assertEquals(0, rgy.getStreamScopes().keySet().size());
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/4b36d473/build.xml
----------------------------------------------------------------------
diff --git a/build.xml b/build.xml
index 93f3fdc..43a72f3 100644
--- a/build.xml
+++ b/build.xml
@@ -112,6 +112,7 @@
<ant dir="runtime/jsoncontrol" target="@{target}" useNativeBasedir="true"/>
<ant dir="runtime/jobregistry" target="@{target}" useNativeBasedir="true"/>
<ant dir="utils/metrics" target="@{target}" useNativeBasedir="true"/>
+ <ant dir="utils/streamscope" target="@{target}" useNativeBasedir="true"/>
<ant dir="providers/direct" target="@{target}" useNativeBasedir="true"/>
<ant dir="connectors/common" target="@{target}" useNativeBasedir="true"/>
@@ -215,6 +216,7 @@
</classfiles>
<sourcefiles encoding="UTF-8">
<fileset dir="${basedir}/utils/metrics/src/main/java" includes="**/*.java"/>
+ <fileset dir="${basedir}/utils/streamscope/src/main/java" includes="**/*.java"/>
</sourcefiles>
</group>
<group name="Quarks Connectors">
@@ -321,7 +323,7 @@
<group title="Quarks Connectors" packages="quarks.connectors.*"/>
<group title="Quarks Samples" packages="quarks.samples.*"/>
<group title="Quarks Analytics" packages="quarks.analytics.*"/>
- <group title="Quarks Utilities" packages="quarks.metrics,quarks.metrics.*"/>
+ <group title="Quarks Utilities" packages="quarks.metrics,quarks.metrics.*,quarks.streamscope,quarks.streamscope.*"/>
<group title="Quarks Low-Level API" packages="quarks.graph,quarks.graph.*,quarks.oplet,quarks.oplet.*,quarks.window"/>
<group title="Quarks SPI" packages="quarks.*.spi.*"/>
</javadoc>
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/4b36d473/providers/development/build.xml
----------------------------------------------------------------------
diff --git a/providers/development/build.xml b/providers/development/build.xml
index df53ecd..a8ea24b 100644
--- a/providers/development/build.xml
+++ b/providers/development/build.xml
@@ -27,12 +27,14 @@
<pathelement location="${quarks.lib}/quarks.providers.direct.jar" />
<pathelement location="${quarks.console}/server/lib/quarks.console.server.jar" />
<pathelement location="${quarks.utils}/metrics/lib/quarks.utils.metrics.jar" />
+ <pathelement location="${quarks.utils}/streamscope/lib/quarks.utils.streamscope.jar" />
<pathelement location="${lib}/quarks.runtime.jmxcontrol.jar" />
</path>
<path id="test.compile.classpath">
<pathelement location="${jar}" />
<pathelement location="../../api/topology/test.classes" />
+ <pathelement location="../../utils/streamscope/test.classes" />
<path refid="compile.classpath"/>
</path>
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/4b36d473/providers/development/src/main/java/quarks/providers/development/DevelopmentProvider.java
----------------------------------------------------------------------
diff --git a/providers/development/src/main/java/quarks/providers/development/DevelopmentProvider.java b/providers/development/src/main/java/quarks/providers/development/DevelopmentProvider.java
index cd2ca9f..a6a4ae7 100644
--- a/providers/development/src/main/java/quarks/providers/development/DevelopmentProvider.java
+++ b/providers/development/src/main/java/quarks/providers/development/DevelopmentProvider.java
@@ -38,9 +38,9 @@ import quarks.oplet.core.FanOut;
import quarks.oplet.core.Peek;
import quarks.providers.direct.DirectProvider;
import quarks.runtime.jmxcontrol.JMXControlService;
+import quarks.streamscope.StreamScope;
+import quarks.streamscope.StreamScopeRegistry;
import quarks.topology.Topology;
-import quarks.topology.plumbing.StreamScope;
-import quarks.topology.plumbing.StreamScopeRegistry;
/**
* Provider intended for development.
@@ -92,8 +92,8 @@ public class DevelopmentProvider extends DirectProvider {
@Override
public Future<Job> submit(Topology topology, JsonObject config) {
- Metrics.counter(topology);
addStreamScopes(topology);
+ Metrics.counter(topology);
duplicateTags(topology);
return super.submit(topology, config);
}
@@ -154,7 +154,7 @@ public class DevelopmentProvider extends DirectProvider {
t.graph().peekAll(
() -> {
StreamScope<?> streamScope = new StreamScope<>();
- Peek<?> peekOp = new quarks.oplet.plumbing.StreamScope<>(streamScope);
+ Peek<?> peekOp = new quarks.streamscope.oplets.StreamScope<>(streamScope);
registerStreamScope(rgy, peekOp, streamScope);
return peekOp;
},
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/4b36d473/providers/development/src/test/java/quarks/test/providers/dev/streamscope/DevelopmentStreamScopeTest.java
----------------------------------------------------------------------
diff --git a/providers/development/src/test/java/quarks/test/providers/dev/streamscope/DevelopmentStreamScopeTest.java b/providers/development/src/test/java/quarks/test/providers/dev/streamscope/DevelopmentStreamScopeTest.java
new file mode 100644
index 0000000..fafe8eb
--- /dev/null
+++ b/providers/development/src/test/java/quarks/test/providers/dev/streamscope/DevelopmentStreamScopeTest.java
@@ -0,0 +1,25 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you 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 quarks.test.providers.dev.streamscope;
+
+import quarks.test.providers.dev.DevelopmentTestSetup;
+import quarks.test.streamscope.StreamScopeTest;
+
+public class DevelopmentStreamScopeTest extends StreamScopeTest implements DevelopmentTestSetup {
+}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/4b36d473/utils/.classpath
----------------------------------------------------------------------
diff --git a/utils/.classpath b/utils/.classpath
index 0eeaf77..1539f57 100644
--- a/utils/.classpath
+++ b/utils/.classpath
@@ -2,6 +2,8 @@
<classpath>
<classpathentry kind="src" path="metrics/src/main/java"/>
<classpathentry kind="src" path="metrics/src/test/java"/>
+ <classpathentry kind="src" path="streamscope/src/main/java"/>
+ <classpathentry kind="src" path="streamscope/src/test/java"/>
<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER"/>
<classpathentry combineaccessrules="false" kind="src" path="/api"/>
<classpathentry combineaccessrules="false" kind="src" path="/ext"/>
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/4b36d473/utils/streamscope/build.xml
----------------------------------------------------------------------
diff --git a/utils/streamscope/build.xml b/utils/streamscope/build.xml
new file mode 100644
index 0000000..a27c6d4
--- /dev/null
+++ b/utils/streamscope/build.xml
@@ -0,0 +1,44 @@
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you 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.
+ -->
+<project name="quarks.utils.streamscope" default="all" xmlns:jacoco="antlib:org.jacoco.ant">
+ <description>
+ Build the StreamScope add-on.
+ </description>
+
+ <property name="component.path" value="utils/streamscope"/>
+ <import file="../../common-build.xml"/>
+
+ <path id="compile.classpath">
+ <pathelement location="${quarks.lib}/quarks.api.topology.jar"/>
+ <path refid="quarks.ext.classpath"/>
+ </path>
+
+ <path id="test.compile.classpath">
+ <pathelement location="${jar}" />
+ <pathelement location="${quarks.lib}/quarks.providers.direct.jar"/>
+ <pathelement location="../../api/topology/test.classes"/>
+ <pathelement location="../../providers/direct/test.classes"/>
+ <path refid="compile.classpath"/>
+ </path>
+
+ <path id="test.classpath">
+ <pathelement location="${test.classes}" />
+ <path refid="test.compile.classpath"/>
+ <path refid="test.common.classpath" />
+ </path>
+
+</project>
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/4b36d473/utils/streamscope/src/main/java/quarks/streamscope/StreamScope.java
----------------------------------------------------------------------
diff --git a/utils/streamscope/src/main/java/quarks/streamscope/StreamScope.java b/utils/streamscope/src/main/java/quarks/streamscope/StreamScope.java
new file mode 100644
index 0000000..17a26ce
--- /dev/null
+++ b/utils/streamscope/src/main/java/quarks/streamscope/StreamScope.java
@@ -0,0 +1,415 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you 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 quarks.streamscope;
+
+import java.util.Collections;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+
+import quarks.function.Consumer;
+import quarks.function.Functions;
+import quarks.function.Predicate;
+import quarks.topology.TStream;
+
+/**
+ * A Stream "oscilloscope" for capturing stream tuples for analysis / debug.
+ * <P>
+ * A {@code StreamScope} is expected to be used as parameter to
+ * {@link TStream#peek(Consumer)}.
+ * </P><P>
+ * A {@link TriggerManager} controls which tuples are captured.
+ * A {@link BufferManager} controls the retention policy for captured tuples.
+ * </P>
+ * <P>
+ * StreamScope instances are typically registered in and located via
+ * a {@link StreamScopeRegistry}.
+ * </P>
+ * @see StreamScopeRegistry
+ * @see quarks.providers.development.DevelopmentProvider DevelopmentProvider
+ */
+public class StreamScope<T> implements Consumer<T> {
+ private static final long serialVersionUID = 1L;
+ private final BufferManager<T> buffer = new BufferManager<>();
+ private final TriggerManager<T> trigger = new TriggerManager<>();
+ private boolean isEnabled;
+
+ /**
+ * A captured tuple.
+ * <P>
+ * The Sample captures the tuple, and the system time and nanoTime
+ * that the tuple was received.
+ * </P>
+ *
+ * @param <T> Tuple type.
+ */
+ public static class Sample<T> {
+ private final long ts;
+ private final long nanoTime;
+ private final T tuple;
+
+ Sample(T tuple) {
+ this.ts = System.currentTimeMillis();
+ this.nanoTime = System.nanoTime();
+ this.tuple = tuple;
+ }
+
+ /**
+ * Capture time in msec since the epoch.
+ * @return the timestamp
+ * @see System#currentTimeMillis()
+ */
+ public long timestamp() {
+ return ts;
+ }
+
+ /**
+ * Capture time in nanoTime.
+ * @return the nanoTime
+ * @see System#nanoTime()
+ */
+ public long nanoTime() {
+ return nanoTime;
+ }
+
+ /**
+ * The captured tuple.
+ * @return the tuple
+ */
+ public T tuple() {
+ return tuple;
+ }
+
+ @Override
+ public String toString() {
+ return "ts="+ts+" nano="+nanoTime+" tuple="+tuple;
+ }
+ }
+
+ /**
+ * Control the retention of captured tuples.
+ * <P>
+ * Captured tuples are retained until either:
+ * <ul>
+ * <li>a maximum retention count is exceeded</li>
+ * <li>TODO a maximum retention time is exceeded</li>
+ * </ul>
+ * </P><P>
+ * The default configuration is a maxCount of 10.
+ * </P>
+ */
+ public static class BufferManager<T> {
+ private List<Sample<T>> buffer = Collections.emptyList();
+ private int maxCount = 10;
+ private long period;
+ private TimeUnit unit;
+
+ // TODO timer based eviction
+ // TODO look at using quarks Windows / WindowImpl for the buffer.
+ // Can window type/"size" be changed?
+ // Does it support simultaneous maxCount & maxAge?
+
+ List<Sample<T>> getSamples() {
+ return Collections.unmodifiableList(buffer);
+ }
+
+ /**
+ * Set the maximum number of tuples to retain.
+ * <P>
+ * The capture buffer is cleared.
+ * </P>
+ * @param maxCount the maximum number of tuples to retain.
+ * Specify 0 to disable count based retention.
+ */
+ public void setMaxRetentionCount(int maxCount) {
+ this.maxCount = maxCount;
+ allocate();
+ }
+
+ /**
+ * Set the maximum retention time of a tuple.
+ * <P>
+ * The capture buffer is cleared.
+ * </P>
+ * @param age the amount of time to retain a tuple.
+ * Specify 0 to disable age based retention.
+ * @param unit {@link TimeUnit}
+ */
+ public void setMaxRetentionTime(long age, TimeUnit unit) {
+ if (age < 0)
+ throw new IllegalArgumentException("age");
+ Objects.requireNonNull(unit, "unit");
+ this.period = age;
+ this.unit = unit;
+
+ throw new IllegalStateException("setMaxRetentionTime is NYI");
+ }
+
+ /**
+ * Get the number of tuples in the capture buffer.
+ * @return the count.
+ */
+ int getCount() {
+ return buffer.size();
+ }
+
+ void release() {
+ buffer = Collections.emptyList();
+ }
+
+ void allocate() {
+ buffer = new LinkedList<>();
+ }
+
+ void add(Sample<T> sample) {
+ if (maxCount > 0 && buffer.size() >= maxCount)
+ buffer.remove(0);
+ buffer.add(sample);
+ }
+
+ @Override
+ public String toString() {
+ return "size="+getCount()+" maxCount="+maxCount+" maxAge="+period+(unit==null ? "" : unit);
+ }
+
+ }
+
+ /**
+ * Control what triggers capturing of tuples.
+ * <P>
+ * The following modes are supported:
+ * <ul>
+ * <li>continuous - capture every tuple</li>
+ * <li>by-count - capture every nth tuple</li>
+ * <li>by-time - capture based on time elapsed since the last capture</li>
+ * <li>by-Predicate - capture based on evaluating a predicate</li>
+ * </ul>
+ * </P><P>
+ * Tuple capture can be temporarily paused via {@link TriggerManager#setPaused(boolean) setPaused}.
+ * Pausing capture does not clear the capture buffer.
+ * </P><P>
+ * Capture processing can be automatically paused when a particular event
+ * has occurred via {@link TriggerManager#setPauseOn(Predicate) setPauseOn}.
+ * The pause predicate is evaluated after processing each received tuple.
+ * Use {@link TriggerManager#setPaused(boolean)} setPaused} to re-enable capture
+ * following a triggered pause.
+ * </P><P>
+ * The default configuration is {@code continuous} and not paused.
+ * </P>
+ */
+ public static class TriggerManager<T> {
+ private Predicate<T> predicate = Functions.alwaysTrue();
+ private Predicate<T> pauseOnPredicate = Functions.alwaysFalse();
+ private boolean paused = false;
+
+ /**
+ * Test if the tuple should be captured.
+ * @param tuple the tuple
+ * @return true to capture the tuple.
+ */
+ boolean test(T tuple) {
+ if (paused)
+ return false;
+ boolean b = predicate.test(tuple);
+ paused = pauseOnPredicate.test(tuple);
+ return b;
+ }
+
+ /**
+ * Set capture paused control
+ * @param paused true to pause, false to clear pause.
+ */
+ public void setPaused(boolean paused) {
+ this.paused = paused;
+ }
+
+ /**
+ * Is capture paused?
+ * @return true if paused
+ */
+ public boolean isPaused() {
+ return paused;
+ }
+
+ /**
+ * Set a pause-on predicate. Capture is paused if the predicate
+ * returns true;
+ * @param predicate the predicate
+ * @see Functions#alwaysFalse()
+ */
+ public void setPauseOn(Predicate<T> predicate) {
+ Objects.requireNonNull(predicate, "predicate");
+ pauseOnPredicate = predicate;
+ }
+
+ /**
+ * Capture every tuple.
+ */
+ public void setContinuous() {
+ setByPredicate(Functions.alwaysTrue());
+ }
+
+ /**
+ * Capture the first and every nth tuple
+ * @param count
+ */
+ public void setByCount(int count) {
+ setByPredicate(newByCountPredicate(count));
+ }
+
+ /**
+ * Capture the 1st tuple and then the next tuple after {@code period}
+ * {@code unit} time has elapsed since the previous captured tuple.
+ *
+ * @param elapsed time to delay until next capture
+ * @param unit {@link TimeUnit}
+ */
+ public void setByTime(long elapsed, TimeUnit unit) {
+ setByPredicate(newByTimePredicate(elapsed, unit));
+ }
+
+ /**
+ * Capture a tuple if the {@code predicate} test of the tuple returns true.
+ * @param predicate
+ */
+ public void setByPredicate(Predicate<T> predicate) {
+ Objects.requireNonNull(predicate, "predicate");
+ this.predicate = predicate;
+ }
+
+ private static <T> Predicate<T> newByCountPredicate(int count) {
+ if (count < 1)
+ throw new IllegalArgumentException("count");
+ return new Predicate<T>() {
+ private static final long serialVersionUID = 1L;
+ int byCount = count;
+ int curCount = -1; // capture 1st and every byCount-nth
+
+ @Override
+ public boolean test(T value) {
+ return ++curCount % byCount == 0;
+ }
+ };
+ }
+
+ private static <T> Predicate<T> newByTimePredicate(long elapsed, TimeUnit unit) {
+ if (elapsed < 1)
+ throw new IllegalArgumentException("elapsed");
+ Objects.requireNonNull(unit, "unit");
+ return new Predicate<T>() {
+ private static final long serialVersionUID = 1L;
+ private long nextTime;
+
+ @Override
+ public boolean test(T value) {
+ long now = System.currentTimeMillis();
+ if (now > nextTime) {
+ nextTime = now + unit.toMillis(elapsed);
+ return true;
+ }
+ return false;
+ }
+ };
+ }
+
+ }
+
+ /**
+ * Create a new instance.
+ * <P>
+ * Sample capture is disabled.
+ * </P>
+ */
+ public StreamScope() {
+ }
+
+ /**
+ * Is tuple capture enabled?
+ * @return true if enabled.
+ */
+ public boolean isEnabled() {
+ return isEnabled;
+ }
+
+ /**
+ * Enable or disable tuple capture.
+ * <P>
+ * Disabling releases the capture buffer.
+ * </P>
+ * @param isEnabled true to enable, false to disable.
+ */
+ public synchronized void setEnabled(boolean isEnabled) {
+ if (!isEnabled)
+ buffer.release();
+ buffer.allocate();
+ this.isEnabled = isEnabled;
+ }
+
+ /**
+ * Get the {@link BufferManager}
+ * @return the manager
+ */
+ public BufferManager<T> bufferMgr() {
+ return buffer;
+ }
+
+ /**
+ * Get the {@link TriggerManager}
+ * @return the manager
+ */
+ public TriggerManager<T> triggerMgr() {
+ return trigger;
+ }
+
+ /**
+ * Get all captured tuples.
+ * <P>
+ * The returned samples are removed from the capture buffer.
+ * </P>
+ * @return unmodifiable list of captured samples
+ */
+ public synchronized List<Sample<T>> getSamples() {
+ List<Sample<T>> tmp = buffer.getSamples();
+ buffer.allocate();
+ return tmp;
+ }
+
+ /**
+ * Get the number of Samples currently captured
+ * @return the count
+ */
+ public synchronized int getSampleCount() {
+ return buffer.getCount();
+ }
+
+ @Override
+ public synchronized void accept(T tuple) {
+ if (!isEnabled())
+ return;
+ if (trigger.test(tuple))
+ buffer.add(new Sample<T>(tuple));
+ }
+
+ @Override
+ public String toString() {
+ return "isEnabled="+isEnabled+" bufferMgr="+bufferMgr();
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/4b36d473/utils/streamscope/src/main/java/quarks/streamscope/StreamScopeRegistry.java
----------------------------------------------------------------------
diff --git a/utils/streamscope/src/main/java/quarks/streamscope/StreamScopeRegistry.java b/utils/streamscope/src/main/java/quarks/streamscope/StreamScopeRegistry.java
new file mode 100644
index 0000000..836428b
--- /dev/null
+++ b/utils/streamscope/src/main/java/quarks/streamscope/StreamScopeRegistry.java
@@ -0,0 +1,145 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you 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 quarks.streamscope;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * A registry for Stream "oscilloscope" {@link StreamScope} instances.
+ * <P>
+ * The registry contains a collection of StreamScope instances
+ * that are registered by one or more names.
+ * </P><P>
+ * The names are: by a TStream {@link quarks.topology.TStream#alias(String) alias} or
+ * by a stream's (output port's) unique identifier.
+ * Static methods are provided for composing these names and extracting
+ * the alias/identifier from generated names.
+ * </P>
+ * @see quarks.providers.development.DevelopmentProvider DevelopmentProvider
+ */
+public class StreamScopeRegistry {
+ private final Map<String, StreamScope<?>> byNameMap = new HashMap<>();
+ private final Map<StreamScope<?>, List<String>> byStreamScopeMap = new HashMap<>();
+
+ public StreamScopeRegistry() {
+
+ }
+
+ /** create a registration name for a stream alias */
+ public static String nameByStreamAlias(String alias) {
+ Objects.requireNonNull(alias, "alias");
+ return "alias."+alias;
+ }
+
+ /** create a registration name for a stream id */
+ public static String nameByStreamId(String id) {
+ Objects.requireNonNull(id, "id");
+ return "id."+id;
+ }
+
+ /** returns null if {@code name} is not a from nameByStreamAlias() */
+ public static String streamAliasFromName(String name) {
+ Objects.requireNonNull(name, "name");
+ if (!name.startsWith("alias."))
+ return null;
+ return name.substring("alias.".length());
+ }
+
+ /** returns null if {@code name} is not a from nameByStreamId() */
+ public static String streamIdFromName(String name) {
+ Objects.requireNonNull(name, "name");
+ if (!name.startsWith("id."))
+ return null;
+ return name.substring("id.".length());
+ }
+
+ /** A single StreamScope can be registered with multiple names.
+ * @throws IllegalStateException if a registration already exists for {@code name}
+ */
+ public synchronized void register(String name, StreamScope<?> streamScope) {
+ if (byNameMap.containsKey(name))
+ throw new IllegalStateException("StreamScope already registered by name "+name);
+ byNameMap.put(name, streamScope);
+ List<String> names = byStreamScopeMap.get(streamScope);
+ if (names == null) {
+ names = new ArrayList<>(2);
+ byStreamScopeMap.put(streamScope, names);
+ }
+ names.add(name);
+ }
+
+ /**
+ * @param name
+ * @return the StreamScope. null if name is not registered.
+ */
+ public synchronized StreamScope<?> lookup(String name) {
+ return byNameMap.get(name);
+ }
+
+ /**
+ * Get the registered names.
+ * @return unmodifiable collection of the name.
+ * The set is backed by the registry so the set may change.
+ */
+ public synchronized Set<String> getNames() {
+ return Collections.unmodifiableSet(byNameMap.keySet());
+ }
+
+ /** Get registered StreamScopes and the name(s) each is registered with.
+ * The map is backed by the registry so its contents may change.
+ */
+ public synchronized Map<StreamScope<?>, List<String>> getStreamScopes() {
+ return Collections.unmodifiableMap(byStreamScopeMap);
+ }
+
+ /** remove the specific name registration. Other registration of the same StreamScope may still exist.
+ * no-op if name is not registered.
+ * @see #unregister(StreamScope)
+ */
+ public synchronized void unregister(String name) {
+ StreamScope<?> streamScope = byNameMap.get(name);
+ if (streamScope == null)
+ return;
+ byNameMap.remove(name);
+ List<String> names = byStreamScopeMap.get(streamScope);
+ names.remove(name);
+ if (names.isEmpty())
+ byStreamScopeMap.remove(streamScope);
+ }
+
+ /** remove all name registrations of the StreamScope.
+ * no-op if no registrations for the StreamScope
+ */
+ public synchronized void unregister(StreamScope<?> streamScope) {
+ List<String> names = byStreamScopeMap.get(streamScope);
+ if (names == null)
+ return;
+ names = new ArrayList<>(names);
+ for (String name : names)
+ unregister(name);
+ }
+
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/4b36d473/utils/streamscope/src/main/java/quarks/streamscope/oplets/StreamScope.java
----------------------------------------------------------------------
diff --git a/utils/streamscope/src/main/java/quarks/streamscope/oplets/StreamScope.java b/utils/streamscope/src/main/java/quarks/streamscope/oplets/StreamScope.java
new file mode 100644
index 0000000..7ecdfd5
--- /dev/null
+++ b/utils/streamscope/src/main/java/quarks/streamscope/oplets/StreamScope.java
@@ -0,0 +1,44 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you 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 quarks.streamscope.oplets;
+
+import quarks.function.Consumer;
+import quarks.oplet.functional.Peek;
+
+/**
+ * A Stream "oscilloscope" oplet.
+ * <P>
+ * TODO remove this? Just added to help the Console specialize its presentation
+ * and/or so user can differentiate a StreamScope from any other random Peek oplet use.
+ * </P>
+ *
+ * @param <T> Type of the tuple.
+ */
+public class StreamScope<T> extends Peek<T> {
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * Create a new instance.
+ * @param streamScope the consumer function
+ */
+ public StreamScope(Consumer<T> streamScope) {
+ super(streamScope);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/4b36d473/utils/streamscope/src/test/java/quarks/test/streamscope/StreamScopeTest.java
----------------------------------------------------------------------
diff --git a/utils/streamscope/src/test/java/quarks/test/streamscope/StreamScopeTest.java b/utils/streamscope/src/test/java/quarks/test/streamscope/StreamScopeTest.java
new file mode 100644
index 0000000..3b7afa7
--- /dev/null
+++ b/utils/streamscope/src/test/java/quarks/test/streamscope/StreamScopeTest.java
@@ -0,0 +1,325 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you 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 quarks.test.streamscope;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Ignore;
+import org.junit.Test;
+
+import quarks.streamscope.StreamScope;
+import quarks.streamscope.StreamScopeRegistry;
+import quarks.streamscope.StreamScope.Sample;
+import quarks.test.topology.TopologyAbstractTest;
+
+@Ignore
+public abstract class StreamScopeTest extends TopologyAbstractTest {
+
+ @Test
+ public void testStreamScopeFn() throws Exception {
+
+ StreamScope<Integer> ss = new StreamScope<>();
+
+ List<Sample<Integer>> samples;
+ Sample<Integer> sample;
+
+ assertFalse(ss.isEnabled());
+ assertNotNull(ss.bufferMgr());
+ assertNotNull(ss.triggerMgr());
+ assertEquals(0, ss.getSampleCount());
+ samples = ss.getSamples();
+ assertNotNull(samples);
+ assertEquals(0, samples.size());
+
+ // ---------------- no capture when not enabled
+ ss.accept(1);
+ assertEquals(0, ss.getSampleCount());
+ assertEquals(0, ss.getSamples().size());
+
+ ss.setEnabled(true);
+
+ // ---------------- single capture
+ // note: getSamples() removes captured tuples
+ ss.accept(100);
+ assertEquals(1, ss.getSampleCount());
+ samples = ss.getSamples();
+ assertEquals(1, samples.size());
+ sample = samples.get(0);
+ assertEquals(100, sample.tuple().intValue());
+ assertTrue(sample.timestamp() != 0);
+ assertTrue(sample.nanoTime() != 0);
+ assertEquals(0, ss.getSampleCount());
+ assertEquals(0, ss.getSamples().size());
+
+ // ---------------- next capture/get; different lists
+ List<Sample<Integer>> savedSamples = samples;
+ ss.accept(101);
+ assertEquals(1, ss.getSampleCount());
+ samples = ss.getSamples();
+ assertEquals(101, samples.get(0).tuple().intValue());
+ assertTrue(samples != savedSamples);
+ assertEquals(0, ss.getSampleCount());
+ assertEquals(0, ss.getSamples().size());
+
+ // ---------------- multi capture
+ ss.accept(100);
+ ss.accept(101);
+ ss.accept(102);
+ assertEquals(3, ss.getSampleCount());
+ samples = ss.getSamples();
+ assertEquals(100, samples.get(0).tuple().intValue());
+ assertEquals(101, samples.get(1).tuple().intValue());
+ assertEquals(102, samples.get(2).tuple().intValue());
+ assertEquals(0, ss.getSampleCount());
+ assertEquals(0, ss.getSamples().size());
+
+ // ---------------- disable => clears capture buffer
+ ss.accept(100);
+ ss.accept(101);
+ ss.accept(102);
+ assertEquals(3, ss.getSampleCount());
+ ss.setEnabled(false);
+ assertEquals(0, ss.getSampleCount());
+ assertEquals(0, ss.getSamples().size());
+
+ ss.setEnabled(true);
+
+ // ---------------- buffer control at the limit (no eviction)
+ ss.bufferMgr().setMaxRetentionCount(3);
+ ss.accept(100);
+ ss.accept(101);
+ ss.accept(102);
+ assertEquals(3, ss.getSampleCount());
+ samples = ss.getSamples();
+ assertEquals(100, samples.get(0).tuple().intValue());
+ assertEquals(101, samples.get(1).tuple().intValue());
+ assertEquals(102, samples.get(2).tuple().intValue());
+ assertEquals(0, ss.getSampleCount());
+ assertEquals(0, ss.getSamples().size());
+
+ // ---------------- buffer control with eviction
+ ss.bufferMgr().setMaxRetentionCount(2);
+ ss.accept(100);
+ ss.accept(101);
+ ss.accept(102);
+ assertEquals(2, ss.getSampleCount());
+ samples = ss.getSamples();
+ assertEquals(101, samples.get(0).tuple().intValue());
+ assertEquals(102, samples.get(1).tuple().intValue());
+ assertEquals(0, ss.getSampleCount());
+ assertEquals(0, ss.getSamples().size());
+ ss.bufferMgr().setMaxRetentionCount(10);
+
+ // ---------------- trigger byCount
+ ss.triggerMgr().setByCount(3);
+ ss.accept(100);
+ ss.accept(101);
+ ss.accept(102);
+ ss.accept(103);
+ ss.accept(104);
+ ss.accept(105);
+ ss.accept(106);
+ assertEquals(3, ss.getSampleCount());
+ samples = ss.getSamples();
+ assertEquals(100, samples.get(0).tuple().intValue());
+ assertEquals(103, samples.get(1).tuple().intValue());
+ assertEquals(106, samples.get(2).tuple().intValue());
+
+ // ---------------- trigger byPredicate
+ ss.triggerMgr().setByPredicate(t -> t % 2 == 0);
+ ss.accept(100);
+ ss.accept(101);
+ ss.accept(102);
+ ss.accept(103);
+ ss.accept(104);
+ assertEquals(3, ss.getSampleCount());
+ samples = ss.getSamples();
+ assertEquals(100, samples.get(0).tuple().intValue());
+ assertEquals(102, samples.get(1).tuple().intValue());
+ assertEquals(104, samples.get(2).tuple().intValue());
+
+ // ---------------- trigger byTime
+ ss.triggerMgr().setByTime(100, TimeUnit.MILLISECONDS);
+ ss.accept(100);
+ ss.accept(101);
+ ss.accept(102);
+ Thread.sleep(110);
+ ss.accept(103);
+ ss.accept(104);
+ ss.accept(105);
+ Thread.sleep(110);
+ ss.accept(106);
+ assertEquals(3, ss.getSampleCount());
+ samples = ss.getSamples();
+ assertEquals(100, samples.get(0).tuple().intValue());
+ assertEquals(103, samples.get(1).tuple().intValue());
+ assertEquals(106, samples.get(2).tuple().intValue());
+
+ // ---------------- trigger continuous
+ ss.triggerMgr().setContinuous();
+ ss.accept(100);
+ ss.accept(101);
+ ss.accept(102);
+ assertEquals(3, ss.getSampleCount());
+ samples = ss.getSamples();
+ assertEquals(100, samples.get(0).tuple().intValue());
+ assertEquals(101, samples.get(1).tuple().intValue());
+ assertEquals(102, samples.get(2).tuple().intValue());
+
+ // ---------------- trigger pause
+ ss.accept(100);
+ ss.accept(101);
+ ss.triggerMgr().setPaused(true);
+ assertTrue(ss.triggerMgr().isPaused());
+ ss.accept(102);
+ ss.accept(103);
+ ss.triggerMgr().setPaused(false);
+ assertFalse(ss.triggerMgr().isPaused());
+ ss.accept(104);
+ assertEquals(3, ss.getSampleCount());
+ samples = ss.getSamples();
+ assertEquals(100, samples.get(0).tuple().intValue());
+ assertEquals(101, samples.get(1).tuple().intValue());
+ assertEquals(104, samples.get(2).tuple().intValue());
+
+ // ---------------- trigger pauseOn
+
+ ss.triggerMgr().setPauseOn(t -> t == 102);
+ ss.accept(100);
+ ss.accept(101);
+ ss.accept(102);
+ ss.accept(103);
+ ss.accept(104);
+ ss.accept(105);
+ assertTrue(ss.triggerMgr().isPaused());
+ assertEquals(3, ss.getSampleCount());
+ samples = ss.getSamples();
+ assertEquals(100, samples.get(0).tuple().intValue());
+ assertEquals(101, samples.get(1).tuple().intValue());
+ assertEquals(102, samples.get(2).tuple().intValue());
+ ss.triggerMgr().setPaused(false);
+ ss.accept(1000);
+ ss.accept(1010);
+ ss.accept(102);
+ ss.accept(1030);
+ assertEquals(3, ss.getSampleCount());
+ samples = ss.getSamples();
+ assertEquals(1000, samples.get(0).tuple().intValue());
+ assertEquals(1010, samples.get(1).tuple().intValue());
+ assertEquals(102, samples.get(2).tuple().intValue());
+
+ }
+
+ @Test
+ public void testStreamScopeRegistry() throws Exception {
+
+ StreamScope<Integer> ss1 = new StreamScope<>();
+ StreamScope<Integer> ss2 = new StreamScope<>();
+
+ StreamScopeRegistry rgy = new StreamScopeRegistry();
+
+ assertNotNull(rgy.getNames());
+ assertEquals(0, rgy.getNames().size());
+ assertNotNull(rgy.getStreamScopes());
+ assertEquals(0, rgy.getStreamScopes().size());
+ assertNull(rgy.lookup("xyzzy"));
+ rgy.unregister("xyzzy");
+ rgy.unregister(ss1);
+
+ // ---------- name generation / parse functions
+ String alias1Name = StreamScopeRegistry.nameByStreamAlias("alias1");
+ assertNotNull(alias1Name);
+ String alias2Name = StreamScopeRegistry.nameByStreamAlias("alias2");
+ assertNotNull(alias2Name);
+ assertFalse(alias1Name.equals(alias2Name));
+ String alias1 = StreamScopeRegistry.streamAliasFromName(alias1Name);
+ assertEquals("alias1", alias1);
+
+ String id1Name = StreamScopeRegistry.nameByStreamId("id1");
+ assertNotNull(id1Name);
+ String id2Name = StreamScopeRegistry.nameByStreamId("id2");
+ assertNotNull(id2Name);
+ assertFalse(id1Name.equals(id2Name));
+ String id1 = StreamScopeRegistry.streamIdFromName(id1Name);
+ assertEquals("id1", id1);
+
+ assertFalse(StreamScopeRegistry.nameByStreamAlias("1")
+ .equals(StreamScopeRegistry.nameByStreamId("1")));
+
+ // ---------- register
+ rgy.register(alias1Name, ss1);
+ rgy.register(alias2Name, ss2);
+ rgy.register(id2Name, ss2);
+
+ // ---------- lookup
+ assertSame(ss1, rgy.lookup(alias1Name));
+ assertSame(ss2, rgy.lookup(alias2Name));
+ assertSame(ss2, rgy.lookup(id2Name));
+
+ // ----------- getNames
+ assertEquals(3, rgy.getNames().size());
+ assertTrue(rgy.getNames().contains(alias1Name));
+ assertFalse(rgy.getNames().contains(id1Name));
+ assertTrue(rgy.getNames().contains(alias2Name));
+ assertTrue(rgy.getNames().contains(id2Name));
+
+ // ----------- getStreamScopes
+ assertEquals(2, rgy.getStreamScopes().keySet().size());
+ assertTrue(rgy.getStreamScopes().keySet().contains(ss1));
+ assertTrue(rgy.getStreamScopes().keySet().contains(ss2));
+ assertEquals(1, rgy.getStreamScopes().get(ss1).size());
+ assertTrue(rgy.getStreamScopes().get(ss1).contains(alias1Name));
+ assertEquals(2, rgy.getStreamScopes().get(ss2).size());
+ assertTrue(rgy.getStreamScopes().get(ss2).contains(alias2Name));
+ assertTrue(rgy.getStreamScopes().get(ss2).contains(id2Name));
+
+ // ---------- unregister
+ rgy.unregister(alias1Name);
+ assertNull(rgy.lookup(alias1Name));
+ assertEquals(2, rgy.getNames().size());
+ assertFalse(rgy.getNames().contains(alias1Name));
+ assertFalse(rgy.getStreamScopes().keySet().contains(ss1));
+ assertTrue(rgy.getStreamScopes().keySet().contains(ss2));
+
+ rgy.unregister(alias2Name);
+ assertNull(rgy.lookup(alias2Name));
+ assertEquals(1, rgy.getNames().size());
+ assertFalse(rgy.getNames().contains(alias2Name));
+ assertTrue(rgy.getStreamScopes().keySet().contains(ss2));
+ assertSame(ss2, rgy.lookup(id2Name));
+ rgy.unregister(id2Name);
+ assertEquals(0, rgy.getNames().size());
+ assertEquals(0, rgy.getStreamScopes().keySet().size());
+
+ rgy.register(alias2Name, ss2);
+ rgy.register(id2Name, ss2);
+ rgy.unregister(ss2);
+ assertEquals(0, rgy.getNames().size());
+ assertEquals(0, rgy.getStreamScopes().keySet().size());
+ }
+
+}