You are viewing a plain text version of this content. The canonical link for it is here.
Posted to s4-commits@incubator.apache.org by mm...@apache.org on 2012/01/03 14:03:28 UTC
[11/50] [abbrv] git commit: Experiment with builder pattern.
Experiment with builder pattern.
Project: http://git-wip-us.apache.org/repos/asf/incubator-s4/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-s4/commit/fb61d6df
Tree: http://git-wip-us.apache.org/repos/asf/incubator-s4/tree/fb61d6df
Diff: http://git-wip-us.apache.org/repos/asf/incubator-s4/diff/fb61d6df
Branch: refs/heads/piper
Commit: fb61d6df62da3be7551ac1bef2db1567be35e4f2
Parents: d6e3684
Author: Leo Neumeyer <le...@s4.io>
Authored: Sun Dec 11 20:10:05 2011 -0800
Committer: Leo Neumeyer <le...@s4.io>
Committed: Sun Dec 11 20:10:05 2011 -0800
----------------------------------------------------------------------
.../java/org/apache/s4/appbuilder/AppMaker.java | 8 +-
.../main/java/org/apache/s4/appbuilder/Main.java | 28 --
.../main/java/org/apache/s4/appbuilder/MyApp.java | 31 ++-
.../java/org/apache/s4/appbuilder/StreamMaker.java | 16 +
.../src/main/java/org/apache/s4/core/App.java | 113 ++++++--
.../main/java/org/apache/s4/core/EventSource.java | 24 +-
.../src/main/java/org/apache/s4/core/Key.java | 1 -
.../main/java/org/apache/s4/core/KeyFinder.java | 3 +-
.../java/org/apache/s4/core/ProcessingElement.java | 222 +++++++--------
.../src/main/java/org/apache/s4/core/Stream.java | 180 +++++++-----
.../main/java/org/apache/s4/core/Streamable.java | 13 +-
.../java/org/apache/s4/core/CountTriggerTest.java | 16 +
.../java/org/apache/s4/core/NoTriggerTest.java | 17 ++
.../java/org/apache/s4/core/TimeTriggerTest.java | 19 ++
.../test/java/org/apache/s4/core/TriggerTest.java | 75 +++++
.../java/org/apache/s4/core/TriggerablePE.java | 80 ++++++
.../test/java/org/apache/s4/core/TriggeredApp.java | 54 ++++
.../java/org/apache/s4/core/TriggeredModule.java | 6 +
.../org/apache/s4/core/apploading/SimpleApp.java | 6 +-
.../apache/s4/core/triggers/CountTriggerTest.java | 16 -
.../org/apache/s4/core/triggers/NoTriggerTest.java | 17 --
.../apache/s4/core/triggers/TimeTriggerTest.java | 19 --
.../org/apache/s4/core/triggers/TriggerTest.java | 75 -----
.../org/apache/s4/core/triggers/TriggerablePE.java | 80 ------
.../org/apache/s4/core/triggers/TriggeredApp.java | 58 ----
.../apache/s4/core/triggers/TriggeredModule.java | 6 -
.../java/org/apache/s4/wordcount/WordCountApp.java | 10 +-
.../apache/s4/example/counter/CountKeyFinder.java | 1 -
.../java/org/apache/s4/example/counter/MyApp.java | 74 +++---
.../org/apache/s4/example/model/Controller.java | 51 ++--
.../java/org/apache/s4/example/model/MyApp.java | 42 ++--
31 files changed, 713 insertions(+), 648 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/fb61d6df/subprojects/s4-core/src/main/java/org/apache/s4/appbuilder/AppMaker.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/appbuilder/AppMaker.java b/subprojects/s4-core/src/main/java/org/apache/s4/appbuilder/AppMaker.java
index f688809..dcee431 100644
--- a/subprojects/s4-core/src/main/java/org/apache/s4/appbuilder/AppMaker.java
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/appbuilder/AppMaker.java
@@ -11,7 +11,7 @@ import com.google.common.collect.LinkedListMultimap;
import com.google.common.collect.Maps;
import com.google.common.collect.Multimap;
-abstract public class AppMaker {
+abstract public class AppMaker extends App {
/**
* NOTES: reflection+guice:
@@ -38,12 +38,6 @@ abstract public class AppMaker {
private Multimap<PEMaker, StreamMaker> psGraph = LinkedListMultimap.create();
private Map<StreamMaker, PEMaker> spGraph = Maps.newHashMap();
- public AppMaker() {
-
- }
-
- abstract protected void define();
-
void add(PEMaker pem, StreamMaker stream) {
psGraph.put(pem, stream);
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/fb61d6df/subprojects/s4-core/src/main/java/org/apache/s4/appbuilder/Main.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/appbuilder/Main.java b/subprojects/s4-core/src/main/java/org/apache/s4/appbuilder/Main.java
deleted file mode 100644
index 7f7cbab..0000000
--- a/subprojects/s4-core/src/main/java/org/apache/s4/appbuilder/Main.java
+++ /dev/null
@@ -1,28 +0,0 @@
-package org.apache.s4.appbuilder;
-
-public class Main {
-
- /**
- * @param args
- */
- public static void main(String[] args) {
-
- AppMaker am = new AppMaker();
-
- PEMaker pem1, pem2;
- StreamMaker s1;
- StreamMaker s2, s3;
-
- pem1 = am.addPE(PEZ.class);
-
- s1 = am.addStream(EventA.class).withName("My first stream.").withKeyFinder("{gender}").to(pem1);
-
- pem2 = am.addPE(PEY.class).to(s1);
-
- s2 = am.addStream(EventB.class).withName("My second stream.").withKeyFinder("{age}").to(pem2);
-
- s3 = am.addStream(EventB.class).withName("My third stream.").withKeyFinder("{height}").to(pem2);
-
- am.addPE(PEX.class).to(s2).to(s3);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/fb61d6df/subprojects/s4-core/src/main/java/org/apache/s4/appbuilder/MyApp.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/appbuilder/MyApp.java b/subprojects/s4-core/src/main/java/org/apache/s4/appbuilder/MyApp.java
index f948cc7..5f3af49 100644
--- a/subprojects/s4-core/src/main/java/org/apache/s4/appbuilder/MyApp.java
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/appbuilder/MyApp.java
@@ -2,11 +2,21 @@ package org.apache.s4.appbuilder;
public class MyApp extends AppMaker {
- /**
- * @param args
- */
- public void define() {
+ public static void main(String[] args) {
+
+ MyApp myApp = new MyApp();
+ myApp.init();
+ System.out.println(myApp.toString());
+ }
+
+ @Override
+ protected void start() {
+ // TODO Auto-generated method stub
+
+ }
+ @Override
+ protected void onInit() {
PEMaker pem1, pem2;
StreamMaker s1;
StreamMaker s2, s3;
@@ -24,10 +34,15 @@ public class MyApp extends AppMaker {
addPE(PEX.class).to(s2).to(s3);
}
- public static void main(String[] args) {
+ @Override
+ protected void onStart() {
+ // TODO Auto-generated method stub
+
+ }
+
+ @Override
+ protected void onClose() {
+ // TODO Auto-generated method stub
- MyApp myApp = new MyApp();
- myApp.define();
- System.out.println(myApp.toString());
}
}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/fb61d6df/subprojects/s4-core/src/main/java/org/apache/s4/appbuilder/StreamMaker.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/appbuilder/StreamMaker.java b/subprojects/s4-core/src/main/java/org/apache/s4/appbuilder/StreamMaker.java
index e84c282..596634b 100644
--- a/subprojects/s4-core/src/main/java/org/apache/s4/appbuilder/StreamMaker.java
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/appbuilder/StreamMaker.java
@@ -2,6 +2,8 @@ package org.apache.s4.appbuilder;
import org.apache.s4.base.Event;
import org.apache.s4.core.KeyFinder;
+import org.apache.s4.core.ProcessingElement;
+import org.apache.s4.core.Stream;
public class StreamMaker {
@@ -69,4 +71,18 @@ public class StreamMaker {
return this;
}
+ private Stream<? extends Event> getStream() {
+
+ // Stream stream = new Stream(appMaker, name, getKeyFinder(), getProcessingElements());
+ return null;// stream;
+ }
+
+ private KeyFinder<?> getKeyFinder() {
+
+ return null;
+ }
+
+ private ProcessingElement[] getProcessingElements() {
+ return null;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/fb61d6df/subprojects/s4-core/src/main/java/org/apache/s4/core/App.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/core/App.java b/subprojects/s4-core/src/main/java/org/apache/s4/core/App.java
index fbb1cc4..9c4ec81 100644
--- a/subprojects/s4-core/src/main/java/org/apache/s4/core/App.java
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/core/App.java
@@ -15,15 +15,17 @@
*/
package org.apache.s4.core;
-import java.util.ArrayList;
import java.util.Arrays;
-import java.util.List;
+import java.util.Collection;
+import java.util.Map;
import java.util.concurrent.TimeUnit;
import org.apache.s4.base.Event;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.google.common.collect.LinkedListMultimap;
+import com.google.common.collect.Multimap;
import com.google.inject.AbstractModule;
import com.google.inject.Guice;
import com.google.inject.Inject;
@@ -36,8 +38,12 @@ public abstract class App extends AbstractModule {
private static final Logger logger = LoggerFactory.getLogger(App.class);
- final private List<ProcessingElement> pePrototypes = new ArrayList<ProcessingElement>();
- final private List<Streamable<? extends Event>> streams = new ArrayList<Streamable<? extends Event>>();
+ /* PE prototype to streams relations. */
+ final private Multimap<ProcessingElement, Streamable<? extends Event>> pe2stream = LinkedListMultimap.create();
+
+ /* Stream prototype to PE relations. */
+ final private Multimap<Streamable<? extends Event>, ProcessingElement> stream2pe = LinkedListMultimap.create();
+
private ClockType clockType = ClockType.WALL_CLOCK;
private int id = -1;
@Inject
@@ -79,55 +85,82 @@ public abstract class App extends AbstractModule {
}
/**
- * @return the pePrototypes
+ * @return all the pePrototypes
+ */
+ Collection<ProcessingElement> getPePrototypes() {
+ return pe2stream.keySet();
+ }
+
+ /**
+ * @return all the pePrototypes
*/
- public List<ProcessingElement> getPePrototypes() {
- return pePrototypes;
+ <T extends Event> Collection<ProcessingElement> getTargetPEs(Stream<T> stream) {
+
+ Map<Streamable<?>, Collection<ProcessingElement>> stream2peMap = stream2pe.asMap();
+
+ return stream2peMap.get(stream);
}
- protected abstract void start();
+ protected abstract void onStart();
- protected abstract void init();
+ protected void start() {
+
+ /* Start all streams. */
+ for (Streamable<? extends Event> stream : getStreams()) {
+ stream.start();
+ }
- protected abstract void close();
+ onStart();
+ }
+
+ protected abstract void onInit();
+
+ protected void init() {
+
+ onInit();
+ }
+
+ protected abstract void onClose();
+
+ protected void close() {
+
+ onClose();
+ removeAll();
+ }
- public void removeAll() {
+ private void removeAll() {
- for (ProcessingElement pe : pePrototypes) {
+ for (ProcessingElement pe : getPePrototypes()) {
/* Remove all instances. */
pe.removeAll();
}
- for (Streamable<? extends Event> stream : streams) {
-
- /* Close all streams. */
+ /* Get the set of streams and close them. */
+ for (Streamable<?> stream : getStreams()) {
stream.close();
}
- /* Give prototype a chance to clean up after itself. */
- close();
-
- /* Finally remove from App. */
- pePrototypes.clear();
- streams.clear();
+ /* Finally remove the entire app graph. */
+ pe2stream.clear();
+ stream2pe.clear();
}
- void addPEPrototype(ProcessingElement pePrototype) {
+ void addPEPrototype(ProcessingElement pePrototype, Stream<? extends Event> stream) {
- pePrototypes.add(pePrototype);
+ pe2stream.put(pePrototype, stream);
}
- void addStream(Streamable<? extends Event> stream) {
+ void addStream(Streamable<? extends Event> stream, ProcessingElement pePrototype) {
- streams.add(stream);
+ stream2pe.put(stream, pePrototype);
}
- public List<Streamable<? extends Event>> getStreams() {
- return streams;
+ Collection<Streamable<? extends Event>> getStreams() {
+ return stream2pe.keySet();
}
/**
@@ -217,7 +250,7 @@ public abstract class App extends AbstractModule {
protected <T extends Event> Stream<T> createStream(String name, KeyFinder<T> finder,
ProcessingElement... processingElements) {
- return new Stream<T>(this, name, finder, processingElements);
+ return new Stream<T>(this).withName(name).withKey(finder).to(processingElements);
}
/**
@@ -234,7 +267,28 @@ public abstract class App extends AbstractModule {
*/
protected <T extends Event> Stream<T> createStream(String name, ProcessingElement... processingElements) {
- return new Stream<T>(this, name, processingElements);
+ return new Stream<T>(this).withName(name).to(processingElements);
+ }
+
+ /**
+ * Creates stream with default values. Use the builder methods to configure the stream. Example:
+ * <p>
+ *
+ * <pre>
+ * s1 = <SampleEvent> createStream().withName("My first stream.").withKey(new AKeyFinder()).to(somePE);
+ * </pre>
+ *
+ * <p>
+ *
+ * @param name
+ * the name of the stream
+ * @param processingElements
+ * the target processing elements
+ * @return the stream
+ */
+ protected <T extends Event> Stream<T> createStream(Class<T> type) {
+
+ return new Stream<T>(this);
}
/**
@@ -308,4 +362,5 @@ public abstract class App extends AbstractModule {
// TODO Auto-generated method stub
}
+
}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/fb61d6df/subprojects/s4-core/src/main/java/org/apache/s4/core/EventSource.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/core/EventSource.java b/subprojects/s4-core/src/main/java/org/apache/s4/core/EventSource.java
index b72f34e..21ea23b 100644
--- a/subprojects/s4-core/src/main/java/org/apache/s4/core/EventSource.java
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/core/EventSource.java
@@ -14,16 +14,16 @@ import org.slf4j.LoggerFactory;
* event stream. Each EventSource may have an unlimited number of subscribers.
*
*/
-public class EventSource<T extends Event> implements Streamable<T> {
+class EventSource<T extends Event> extends Streamable<T> {
/* No need to synchronize this object because we expect a single thread. */
private Set<Stream<T>> streams = new HashSet<Stream<T>>();
private static final Logger logger = LoggerFactory.getLogger(EventSource.class);
final private String name;
- public EventSource(App app, String name) {
+ EventSource(App app, String name) {
this.name = name;
- app.addStream(this);
+ app.addStream(this, null);
}
/**
@@ -31,7 +31,7 @@ public class EventSource<T extends Event> implements Streamable<T> {
*
* @param stream
*/
- public void subscribeStream(Stream<T> stream) {
+ void subscribeStream(Stream<T> stream) {
logger.info("Subscribing stream: {} to event source: {}.", stream.getName(), getName());
streams.add(stream);
}
@@ -41,7 +41,7 @@ public class EventSource<T extends Event> implements Streamable<T> {
*
* @param stream
*/
- public void unsubscribeStream(Stream<T> stream) {
+ void unsubscribeStream(Stream<T> stream) {
logger.info("Unsubsubscribing stream: {} to event source: {}.", stream.getName(), getName());
streams.remove(stream);
}
@@ -62,14 +62,14 @@ public class EventSource<T extends Event> implements Streamable<T> {
*
* @return the number of streams subscribed to this event source.
*/
- public int getNumSubscribers() {
+ int getNumSubscribers() {
return streams.size();
}
/**
* @return the name of this event source.
*/
- public String getName() {
+ String getName() {
return name;
}
@@ -77,7 +77,7 @@ public class EventSource<T extends Event> implements Streamable<T> {
* Close all the streams subscribed to this event source.
*/
@Override
- public void close() {
+ void close() {
for (Stream<T> stream : streams) {
logger.info("Closing stream: {} in event source: {}.", stream.getName(), getName());
stream.close();
@@ -88,7 +88,13 @@ public class EventSource<T extends Event> implements Streamable<T> {
*
* @return the set of streams subscribed to this event source.
*/
- public Set<Stream<T>> getStreams() {
+ Set<Stream<T>> getStreams() {
return streams;
}
+
+ @Override
+ void start() {
+ // TODO Auto-generated method stub
+
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/fb61d6df/subprojects/s4-core/src/main/java/org/apache/s4/core/Key.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/core/Key.java b/subprojects/s4-core/src/main/java/org/apache/s4/core/Key.java
index a9bd17d..16dea3c 100644
--- a/subprojects/s4-core/src/main/java/org/apache/s4/core/Key.java
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/core/Key.java
@@ -15,7 +15,6 @@
*/
package org.apache.s4.core;
-
import java.util.List;
import org.apache.commons.lang.StringUtils;
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/fb61d6df/subprojects/s4-core/src/main/java/org/apache/s4/core/KeyFinder.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/core/KeyFinder.java b/subprojects/s4-core/src/main/java/org/apache/s4/core/KeyFinder.java
index cab5646..504e09f 100644
--- a/subprojects/s4-core/src/main/java/org/apache/s4/core/KeyFinder.java
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/core/KeyFinder.java
@@ -15,7 +15,6 @@
*/
package org.apache.s4.core;
-
import java.util.List;
import org.apache.s4.base.Event;
@@ -25,7 +24,7 @@ import org.apache.s4.base.Event;
* (One value for single keys, and more than one value for
* composite keys.)
* */
-public interface KeyFinder<T extends Event> {
+abstract public interface KeyFinder<T extends Event> {
public List<String> get(T event);
}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/fb61d6df/subprojects/s4-core/src/main/java/org/apache/s4/core/ProcessingElement.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/core/ProcessingElement.java b/subprojects/s4-core/src/main/java/org/apache/s4/core/ProcessingElement.java
index 1621c04..5314824 100644
--- a/subprojects/s4-core/src/main/java/org/apache/s4/core/ProcessingElement.java
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/core/ProcessingElement.java
@@ -17,7 +17,6 @@ package org.apache.s4.core;
import java.util.Collection;
import java.util.Map;
-
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ExecutionException;
@@ -31,61 +30,49 @@ import org.apache.s4.core.gen.OverloadDispatcherGenerator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.collect.MapMaker;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
-import com.google.common.collect.Maps;
import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.MapMaker;
+import com.google.common.collect.Maps;
/**
* @author Leo Neumeyer
* @author Matthieu Morel
* <p>
- * Base class for implementing processing in S4. All instances are
- * organized as follows:
+ * Base class for implementing processing in S4. All instances are organized as follows:
* <ul>
- * <li>A PE prototype is a special type of instance that, along with
- * {@link Stream} defines the topology of the application graph.
- * <li>PE prototypes manage the creation and destruction of PE
- * instances.
+ * <li>A PE prototype is a special type of instance that, along with {@link Stream} defines the topology of the
+ * application graph.
+ * <li>PE prototypes manage the creation and destruction of PE instances.
* <li>All PE instances are clones of a PE prototype.
* <li>PE instances are associated with a unique key.
- * <li>PE instances do the actual work by processing any number of input
- * events of various types and emit output events of various types.
- * <li>To process events, {@code ProcessingElement} dynamically matches
- * an event type to a processing method. See
- * {@link org.apache.s4.core.gen.OverloadDispatcher} . There are two
- * types of processing methods:
+ * <li>PE instances do the actual work by processing any number of input events of various types and emit output
+ * events of various types.
+ * <li>To process events, {@code ProcessingElement} dynamically matches an event type to a processing method.
+ * See {@link org.apache.s4.core.gen.OverloadDispatcher} . There are two types of processing methods:
* <ul>
- * <li>{@code onEvent(SomeEvent event)} When implemented, input events
- * of type {@code SomeEvent} will be dispatched to this method.
- * <li>{@code onTrigger(AnotherEvent event)} When implemented, input
- * events of type {@code AnotherEvent} will be dispatched to this method
- * when certain conditions are met. See
+ * <li>{@code onEvent(SomeEvent event)} When implemented, input events of type {@code SomeEvent} will be
+ * dispatched to this method.
+ * <li>{@code onTrigger(AnotherEvent event)} When implemented, input events of type {@code AnotherEvent} will be
+ * dispatched to this method when certain conditions are met. See
* {@link #setTrigger(Class, int, long, TimeUnit)}.
* </ul>
* <li>
- * A PE implementation must not create threads. A periodic task can be
- * implemented by overloading the {@link #onTime()} method. See
- * {@link #setTimerInterval(long, TimeUnit)}
- * <li>If a reference in the PE prototype shared by the PE instances,
- * the object must be thread safe.
- * <li>The code in a PE instance is synchronized by the framework to
- * avoid concurrency problems.
- * <li>In some special cases, it may be desirable to allow concurrency
- * in the PE instance. For example, there may be several event
- * processing methods that can safely run concurrently. To enable
- * concurrency, annotate the implementation of {@code ProcessingElement}
- * with {@link ThreadSafe}.
- * <li>PE instances never use the constructor. They must be initialized
- * by implementing the {@link #onCreate()} method.
- * <li>PE class fields are cloned from the prototype. References are
- * also copied which means that if the prototype creates a collection
- * object, all instances will be sharing the same collection object
- * which is usually <em>NOT</em> what the programmer intended . The
- * application developer is responsible for initializing objects in the
- * {@link #onCreate()} method. For example, if each instance requires a
+ * A PE implementation must not create threads. A periodic task can be implemented by overloading the
+ * {@link #onTime()} method. See {@link #setTimerInterval(long, TimeUnit)}
+ * <li>If a reference in the PE prototype shared by the PE instances, the object must be thread safe.
+ * <li>The code in a PE instance is synchronized by the framework to avoid concurrency problems.
+ * <li>In some special cases, it may be desirable to allow concurrency in the PE instance. For example, there
+ * may be several event processing methods that can safely run concurrently. To enable concurrency, annotate the
+ * implementation of {@code ProcessingElement} with {@link ThreadSafe}.
+ * <li>PE instances never use the constructor. They must be initialized by implementing the {@link #onCreate()}
+ * method.
+ * <li>PE class fields are cloned from the prototype. References are also copied which means that if the
+ * prototype creates a collection object, all instances will be sharing the same collection object which is
+ * usually <em>NOT</em> what the programmer intended . The application developer is responsible for initializing
+ * objects in the {@link #onCreate()} method. For example, if each instance requires a
* <tt>List<tt/> object the PE should implement the following:
* <pre>
* {@code
@@ -117,8 +104,8 @@ public abstract class ProcessingElement implements Cloneable {
protected App app;
/*
- * This maps holds all the instances. We make it package private to prevent
- * concrete classes from updating the collection.
+ * This maps holds all the instances. We make it package private to prevent concrete classes from updating the
+ * collection.
*/
Cache<String, ProcessingElement> peInstances;
@@ -143,8 +130,7 @@ public abstract class ProcessingElement implements Cloneable {
}
/**
- * Create a PE prototype. By default, PE instances will never expire. Use
- * {@code #configurePECache} to configure.
+ * Create a PE prototype. By default, PE instances will never expire. Use {@code #configurePECache} to configure.
*
* @param app
* the app that contains this PE
@@ -159,7 +145,7 @@ public abstract class ProcessingElement implements Cloneable {
}
this.app = app;
- app.addPEPrototype(this);
+ app.addPEPrototype(this, null);
peInstances = CacheBuilder.newBuilder().build(new CacheLoader<String, ProcessingElement>() {
@Override
@@ -171,16 +157,15 @@ public abstract class ProcessingElement implements Cloneable {
triggers = new MapMaker().makeMap();
/*
- * Only the PE Prototype uses the constructor. The PEPrototype field
- * will be cloned by the instances and point to the prototype.
+ * Only the PE Prototype uses the constructor. The PEPrototype field will be cloned by the instances and point
+ * to the prototype.
*/
this.pePrototype = this;
}
/**
- * This method is called by the PE timer. By default it is synchronized with
- * the {@link #onEvent()} and {@link #onTrigger()} methods. To execute
- * concurrently with other methods, the {@link ProcessingElelment} subclass
+ * This method is called by the PE timer. By default it is synchronized with the {@link #onEvent()} and
+ * {@link #onTrigger()} methods. To execute concurrently with other methods, the {@link ProcessingElelment} subclass
* must be annotated with {@link @ThreadSafe}.
*
* Override this method to implement a periodic process.
@@ -189,16 +174,14 @@ public abstract class ProcessingElement implements Cloneable {
}
/**
- * This method is called after a PE instance is created. Use it to
- * initialize fields that are PE instance specific. PE instances are created
- * using {#clone()}. Fields initialized in the class constructor are shared
- * by all PE instances.
+ * This method is called after a PE instance is created. Use it to initialize fields that are PE instance specific.
+ * PE instances are created using {#clone()}. Fields initialized in the class constructor are shared by all PE
+ * instances.
*/
abstract protected void onCreate();
/**
- * This method is called before a PE instance is removed. Use it to close
- * resources and clean up.
+ * This method is called before a PE instance is removed. Use it to close resources and clean up.
*/
abstract protected void onRemove();
@@ -224,11 +207,10 @@ public abstract class ProcessingElement implements Cloneable {
/**
* Set PE expiration and cache size.
* <p>
- * PE instances will be automatically removed from the cache once a fixed
- * duration has elapsed after the PEs creation, or last access.
+ * PE instances will be automatically removed from the cache once a fixed duration has elapsed after the PEs
+ * creation, or last access.
* <p>
- * Least accessed PEs will automatically be removed from the cache when the
- * number of PEs approaches maximumSize.
+ * Least accessed PEs will automatically be removed from the cache when the number of PEs approaches maximumSize.
* <p>
* When this method is called all existing PE instances are destroyed.
*
@@ -239,11 +221,14 @@ public abstract class ProcessingElement implements Cloneable {
* the PE duration
* @param timeUnit
* the time unit
+ * @return the PE prototype
*/
- public void configurePECache(int maximumSize, long duration, TimeUnit timeUnit) {
+ public ProcessingElement withPECache(int maximumSize, long duration, TimeUnit timeUnit) {
- if (!isPrototype)
- return;
+ if (!isPrototype) {
+ logger.error("This method can only be used on the PE prototype. Cache not configured.");
+ return this;
+ }
peInstances = CacheBuilder.newBuilder().expireAfterAccess(duration, timeUnit).maximumSize(maximumSize)
.build(new CacheLoader<String, ProcessingElement>() {
@@ -252,6 +237,8 @@ public abstract class ProcessingElement implements Cloneable {
return createPE(key);
}
});
+
+ return this;
}
/**
@@ -259,42 +246,41 @@ public abstract class ProcessingElement implements Cloneable {
*
* <ul>
* <li>An event of eventType arrived to the PE instance
- * <li>numEvents have arrived since the last time this trigger was fired
- * -OR- time since last event is greater than interval.
+ * <li>numEvents have arrived since the last time this trigger was fired -OR- time since last event is greater than
+ * interval.
* </ul>
*
* <p>
- * When the trigger fires, the method <tt>trigger(EventType event)</tt> is
- * called. Where <tt>EventType</tt> matches the argument eventType.
+ * When the trigger fires, the method <tt>trigger(EventType event)</tt> is called. Where <tt>EventType</tt> matches
+ * the argument eventType.
*
* @param eventType
* the type of event on which this trigger will fire.
* @param numEvents
- * number of events since last trigger activation. Must be
- * greater than zero. (Set to one to trigger on every input
- * event.)
+ * number of events since last trigger activation. Must be greater than zero. (Set to one to trigger on
+ * every input event.)
* @param interval
- * minimum time between triggers. Set to zero if no time interval
- * needed.
+ * minimum time between triggers. Set to zero if no time interval needed.
* @param timeUnit
- * the TimeUnit for the argument interval. Can set to null if no
- * time interval needed.
+ * the TimeUnit for the argument interval. Can set to null if no time interval needed.
+ * @return the PE prototype
*/
- public void setTrigger(Class<? extends Event> eventType, int numEvents, long interval, TimeUnit timeUnit) {
+ public ProcessingElement withTrigger(Class<? extends Event> eventType, int numEvents, long interval,
+ TimeUnit timeUnit) {
if (!isPrototype) {
- logger.warn("This method can only be used on the PE prototype.");
- return;
+ logger.error("This method can only be used on the PE prototype. Trigger not set.");
+ return this;
}
if (eventType == null) {
- logger.error("Argument null in setTrigger() method is not valid.");
- return;
+ logger.error("Argument null in setTrigger() method is not valid. Trigger not set.");
+ return this;
}
if (numEvents < 1) {
- logger.error("Argument numEvents in setTrigger() method must be greater than zero.");
- return;
+ logger.error("Argument numEvents in setTrigger() method must be greater than zero. Trigger not set.");
+ return this;
}
/* Skip trigger checking overhead if there are no triggers. */
@@ -302,12 +288,13 @@ public abstract class ProcessingElement implements Cloneable {
long intervalInMilliseconds = 0;
if (timeUnit != null)
- intervalInMilliseconds = timeUnit.convert(interval,
- TimeUnit.MILLISECONDS);
+ intervalInMilliseconds = timeUnit.convert(interval, TimeUnit.MILLISECONDS);
Trigger config = new Trigger(numEvents, intervalInMilliseconds);
triggers.put(eventType, config);
+
+ return this;
}
/**
@@ -315,6 +302,7 @@ public abstract class ProcessingElement implements Cloneable {
*
* @param timeUnit
* the timeUnt of the returned value.
+ * @return the timer interval.
*/
public long getTimerInterval(TimeUnit timeUnit) {
return timeUnit.convert(timerIntervalInMilliseconds, TimeUnit.MILLISECONDS);
@@ -329,14 +317,15 @@ public abstract class ProcessingElement implements Cloneable {
* in timeUnit
* @param timeUnit
* the timeUnit of interval
+ * @return the PE prototype
*/
- public void setTimerInterval(long interval, TimeUnit timeUnit) {
+ public ProcessingElement withTimerInterval(long interval, TimeUnit timeUnit) {
timerIntervalInMilliseconds = TimeUnit.MILLISECONDS.convert(interval, timeUnit);
/* We only allow timers in the PE prototype, not in the instances. */
if (!isPrototype) {
- logger.warn("This method can only be used on the PE prototype.");
- return;
+ logger.warn("This method can only be used on the PE prototype. Timer not set.");
+ return this;
}
if (timer != null || interval == 0)
@@ -344,16 +333,16 @@ public abstract class ProcessingElement implements Cloneable {
timer = new Timer();
timer.schedule(new OnTimeTask(), 0, timerIntervalInMilliseconds);
+
+ return this;
}
/**
- * Set to true if the concrete PE class has the {@link ThreadSafe}
- * annotation. The default is false (no annotation). In general, application
- * developers don't need to worry about thread safety in the concrete PEs.
- * In some cases the PE needs to be thread safe to avoid deadlocks. For
- * example , if the application graph has cycles and the queues are allowed
- * to block, then some critical PEs with multiple incoming streams need to
- * be made thread safe to avoid locking the entire PE instance.
+ * Set to true if the concrete PE class has the {@link ThreadSafe} annotation. The default is false (no annotation).
+ * In general, application developers don't need to worry about thread safety in the concrete PEs. In some cases the
+ * PE needs to be thread safe to avoid deadlocks. For example , if the application graph has cycles and the queues
+ * are allowed to block, then some critical PEs with multiple incoming streams need to be made thread safe to avoid
+ * locking the entire PE instance.
*
* @return true if the PE implementation is considered thread safe.
*/
@@ -387,13 +376,10 @@ public abstract class ProcessingElement implements Cloneable {
}
/**
- * Checks the trigger for this event type.
- * Creates an inactive trigger if no trigger is found after recursively exploring
- * the event class hierarchy.
- * An inactive trigger never triggers.
+ * Checks the trigger for this event type. Creates an inactive trigger if no trigger is found after recursively
+ * exploring the event class hierarchy. An inactive trigger never triggers.
*
- * @return true if trigger is reached,
- * false if trigger is not ready yet or if trigger is inactive
+ * @return true if trigger is reached, false if trigger is not ready yet or if trigger is inactive
*
*/
private boolean isTrigger(Event event, Class<?> triggerClass) {
@@ -457,9 +443,8 @@ public abstract class ProcessingElement implements Cloneable {
}
/**
- * This method is designed to be used within the package. We make it
- * package-private. The returned instances are all in the same JVM. Do not
- * use it to access remote objects.
+ * This method is designed to be used within the package. We make it package-private. The returned instances are all
+ * in the same JVM. Do not use it to access remote objects.
*/
public ProcessingElement getInstanceForKey(String id) {
@@ -474,8 +459,7 @@ public abstract class ProcessingElement implements Cloneable {
}
/**
- * Get all the local instances. See notes in
- * {@link #getInstanceForKey(String) getLocalInstanceForKey}
+ * Get all the local instances. See notes in {@link #getInstanceForKey(String) getLocalInstanceForKey}
*/
public Collection<ProcessingElement> getInstances() {
@@ -483,9 +467,8 @@ public abstract class ProcessingElement implements Cloneable {
}
/**
- * This method returns a remote PE instance for key. TODO: not implemented
- * for cluster configuration yet, use it only in single node configuration.
- * for testing apps.
+ * This method returns a remote PE instance for key. TODO: not implemented for cluster configuration yet, use it
+ * only in single node configuration. for testing apps.
*
* @return pe instance for key. Null if if doesn't exist.
*/
@@ -499,11 +482,9 @@ public abstract class ProcessingElement implements Cloneable {
}
/**
- * This method returns an immutable map that contains all the PE instances
- * for this prototype. PE instances may be located anywhere in the cluster.
- * Be aware that this could be an expensive operation. TODO: not implemented
- * for cluster configuration yet, use it only in single node configuration.
- * for testing apps.
+ * This method returns an immutable map that contains all the PE instances for this prototype. PE instances may be
+ * located anywhere in the cluster. Be aware that this could be an expensive operation. TODO: not implemented for
+ * cluster configuration yet, use it only in single node configuration. for testing apps.
*/
public Map<String, ProcessingElement> getRemoteInstances() {
@@ -512,16 +493,15 @@ public abstract class ProcessingElement implements Cloneable {
+ "transparently for remote objects once it is implemented.");
/*
- * For now we just return a copy as a placeholder. We need to implement
- * a custom map capable of working on an S4 cluster as efficiently as
- * possible.
+ * For now we just return a copy as a placeholder. We need to implement a custom map capable of working on an S4
+ * cluster as efficiently as possible.
*/
return ImmutableMap.copyOf(peInstances.asMap());
}
/*
- * Called when we create the first PE instance. TODO: Would be better to do
- * this as part of the PE lifecycle after PE construction.
+ * Called when we create the first PE instance. TODO: Would be better to do this as part of the PE lifecycle after
+ * PE construction.
*/
private void onCreateInternal(ProcessingElement pe) {
@@ -530,9 +510,8 @@ public abstract class ProcessingElement implements Cloneable {
logger.trace("OnCreateInternal");
/*
- * If PE class has the @ThreadSafe annotation, then we set isThtreadSafe
- * to true in the prototype so all future PE instances inherit the
- * setting.
+ * If PE class has the @ThreadSafe annotation, then we set isThtreadSafe to true in the prototype so all future
+ * PE instances inherit the setting.
*/
if (pe.getClass().isAnnotationPresent(ThreadSafe.class) == true) {
pe.isThreadSafe = true;
@@ -542,8 +521,7 @@ public abstract class ProcessingElement implements Cloneable {
}
/*
- * Each PE instance needs its own triggers map to keep track time lapsed
- * and event count.
+ * Each PE instance needs its own triggers map to keep track time lapsed and event count.
*/
pe.triggers = Maps.newHashMap(triggers);
}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/fb61d6df/subprojects/s4-core/src/main/java/org/apache/s4/core/Stream.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/core/Stream.java b/subprojects/s4-core/src/main/java/org/apache/s4/core/Stream.java
index 86ea5a0..2d0a3a3 100644
--- a/subprojects/s4-core/src/main/java/org/apache/s4/core/Stream.java
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/core/Stream.java
@@ -15,76 +15,73 @@
*/
package org.apache.s4.core;
-
import java.util.Collection;
+import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import org.apache.s4.base.Event;
-import org.apache.s4.core.Receiver;
-import org.apache.s4.core.Sender;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.google.common.collect.Sets;
+
/**
- * {@link Stream} and {@link ProcessingElement} objects represent the links and
- * nodes in the application graph. A stream sends an {@link Event} object to
- * {@link ProcessingElement} instances located anywhere in a cluster.
+ * {@link Stream} and {@link ProcessingElement} objects represent the links and nodes in the application graph. A stream
+ * sends an {@link Event} object to {@link ProcessingElement} instances located anywhere in a cluster.
* <p>
* Once a stream is instantiated, it is immutable.
* <p>
- * To build an application create stream objects using use the
- * {@link StreamFactory} class.
+ * To build an application create stream objects using use the {@link StreamFactory} class.
*/
-public class Stream<T extends Event> implements Runnable, Streamable<T> {
+public class Stream<T extends Event> extends Streamable<T> implements Runnable {
private static final Logger logger = LoggerFactory.getLogger(Stream.class);
final static private String DEFAULT_SEPARATOR = "^";
final static private int CAPACITY = 1000;
private static int idCounter = 0;
- final private String name;
- final private Key<T> key;
- final private ProcessingElement[] targetPEs;
+ private String name = "";
+ private Key<T> key = null;
+ private ProcessingElement[] targetPEs = null;
final private BlockingQueue<T> queue = new ArrayBlockingQueue<T>(CAPACITY);
- final private Thread thread;
+ private Thread thread;
final private Sender sender;
final private Receiver receiver;
final private int id;
final private App app;
/**
- * Send events using a {@link KeyFinder<T>}. The key finder extracts the
- * value of the key which is used to determine the target
- * {@link org.apache.s4.comm.topology.ClusterNode} for an event.
+ * Send events using a {@link KeyFinder<T>}. The key finder extracts the value of the key which is used to determine
+ * the target {@link org.apache.s4.comm.topology.ClusterNode} for an event.
*
* @param app
* we always register streams with the parent application.
* @param name
- * give this stream a meaningful name in the context of your
- * application.
+ * give this stream a meaningful name in the context of your application.
* @param finder
* the finder object to find the value of the key in an event.
* @param processingElements
* the target PE prototypes for this stream.
*/
- public Stream(App app, String name, KeyFinder<T> finder,
- ProcessingElement... processingElements) {
+ public Stream(App app) {
synchronized (Stream.class) {
id = idCounter++;
}
this.app = app;
- app.addStream(this);
- this.name = name;
+ app.addStream(this, null);
- if (finder == null) {
- this.key = null;
- } else {
- this.key = new Key<T>(finder, DEFAULT_SEPARATOR);
- }
this.sender = app.getSender();
this.receiver = app.getReceiver();
- this.targetPEs = processingElements;
+ }
+
+ void start() {
+
+ /* Get target PE prototypes for this stream. Remove null key. */
+ Set<? extends ProcessingElement> pes = Sets.newHashSet(app.getTargetPEs(this));
+ pes.remove(null);
+ targetPEs = new ProcessingElement[pes.size()];
+ pes.toArray(targetPEs);
/* Start streaming. */
thread = new Thread(this, name);
@@ -93,20 +90,73 @@ public class Stream<T extends Event> implements Runnable, Streamable<T> {
}
/**
- * Send events to all available {@link ProcessingElement} instances
- * contained by the {@link ProcessingElement} prototypes passed to this
- * constructor.
+ * Stop and close this stream.
+ */
+ void close() {
+ thread.interrupt();
+ }
+
+ /**
+ * Name the stream.
*
- * @param app
- * we always register streams with the parent application.
* @param name
- * give this stream a meaningful name in the context of your
- * application.
- * @param processingElements
- * the target PE prototypes for this stream.
+ * the stream name, default is an empty string.
+ * @return the stream maker object
+ */
+ public Stream<T> withName(String name) {
+ this.name = name;
+ return this;
+ }
+
+ /**
+ * Define the key finder for this stream.
+ *
+ * @param keyFinder
+ * a function to lookup the value of the key.
+ * @return the stream maker object
+ */
+ public Stream<T> withKey(KeyFinder<T> keyFinder) {
+ this.key = new Key<T>(keyFinder, DEFAULT_SEPARATOR);
+ return this;
+ }
+
+ /**
+ * Define the key finder for this stream using a descriptor.
+ *
+ * @param keyFinderString
+ * a descriptor to lookup the value of the key.
+ * @return the stream maker object
*/
- public Stream(App app, String name, ProcessingElement... processingElements) {
- this(app, name, null, processingElements);
+ public Stream<T> withKey(String keyFinderString) {
+
+ return this;
+ }
+
+ /**
+ * Send events from this stream to a PE.
+ *
+ * @param pe
+ * a target PE.
+ *
+ * @return the stream maker object
+ */
+ public Stream<T> to(ProcessingElement pe) {
+ app.addStream(this, pe);
+ return this;
+ }
+
+ /**
+ * Send events from this stream to various PEs.
+ *
+ * @param pe
+ * a target PE array.
+ *
+ * @return the stream maker object
+ */
+ public Stream<T> to(ProcessingElement[] pes) {
+ for (int i = 0; i < pes.length; i++)
+ app.addStream(this, pes[i]);
+ return this;
}
/**
@@ -120,50 +170,45 @@ public class Stream<T extends Event> implements Runnable, Streamable<T> {
event.setAppId(app.getId());
/*
- * Events may be sent to local or remote partitions or both. The
- * following code implements the logic.
+ * Events may be sent to local or remote partitions or both. The following code implements the logic.
*/
if (key != null) {
/*
- * We send to a specific PE instance using the key but we don't
- * know if the target partition is remote or local. We need to
- * ask the sender.
+ * We send to a specific PE instance using the key but we don't know if the target partition is remote
+ * or local. We need to ask the sender.
*/
if (sender.sendAndCheckIfLocal(key.get(event), event)) {
/*
- * Sender checked and decided that the target is local so we
- * simply put the event in the queue and we save the trip
- * over the network.
+ * Sender checked and decided that the target is local so we simply put the event in the queue and
+ * we save the trip over the network.
*/
queue.put(event);
}
-
+
} else {
/*
- * We are broadcasting this event to all PE instance. In a
- * cluster, we need to send the event to every node. The sender
- * method takes care of the remote partitions an we take care of
- * putting the event into the queue.
+ * We are broadcasting this event to all PE instance. In a cluster, we need to send the event to every
+ * node. The sender method takes care of the remote partitions an we take care of putting the event into
+ * the queue.
*/
sender.sendToRemotePartitions(event);
queue.put(event);
}
} catch (InterruptedException e) {
- e.printStackTrace();
- logger.error(
- "Interrupted while waiting to put an event in the queue: {}.",
- e.getMessage());
+ if (logger.isTraceEnabled()) {
+ e.printStackTrace();
+ }
+ logger.error("Interrupted while waiting to put an event in the queue: {}.", e.getMessage());
System.exit(-1);
}
}
/**
- * Implements the {@link ReceiverListener} interface. The low level
- * {@link Receiver} object call this method when a new {@link Event} is
- * available.
+ * Implements the {@link ReceiverListener} interface. The low level {@link Receiver} object call this method when a
+ * new {@link Event} is available.
*/
@SuppressWarnings("unchecked")
// Need casting because we don't know the concrete event type.
@@ -172,9 +217,7 @@ public class Stream<T extends Event> implements Runnable, Streamable<T> {
queue.put((T) event);
} catch (InterruptedException e) {
e.printStackTrace();
- logger.error(
- "Interrupted while waiting to put an event in the queue: {}.",
- e.getMessage());
+ logger.error("Interrupted while waiting to put an event in the queue: {}.", e.getMessage());
System.exit(-1);
}
}
@@ -215,13 +258,6 @@ public class Stream<T extends Event> implements Runnable, Streamable<T> {
}
/**
- * Stop and close this stream.
- */
- public void close() {
- thread.interrupt();
- }
-
- /**
* @return the sender object
*/
public Sender getSender() {
@@ -251,8 +287,7 @@ public class Stream<T extends Event> implements Runnable, Streamable<T> {
/* STEP 1: find all PE instances. */
- Collection<ProcessingElement> pes = targetPEs[i]
- .getInstances();
+ Collection<ProcessingElement> pes = targetPEs[i].getInstances();
/* STEP 2: iterate and pass event to PE instance. */
for (ProcessingElement pe : pes) {
@@ -265,8 +300,7 @@ public class Stream<T extends Event> implements Runnable, Streamable<T> {
/* We have a key, send to target PE. */
/* STEP 1: find the PE instance for key. */
- ProcessingElement pe = targetPEs[i]
- .getInstanceForKey(key.get(event));
+ ProcessingElement pe = targetPEs[i].getInstanceForKey(key.get(event));
/* STEP 2: pass event to PE instance. */
pe.handleInputEvent(event);
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/fb61d6df/subprojects/s4-core/src/main/java/org/apache/s4/core/Streamable.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/core/Streamable.java b/subprojects/s4-core/src/main/java/org/apache/s4/core/Streamable.java
index 9e6790e..68b4bcf 100644
--- a/subprojects/s4-core/src/main/java/org/apache/s4/core/Streamable.java
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/core/Streamable.java
@@ -7,22 +7,27 @@ import org.apache.s4.base.Event;
*
* @param <T>
*/
-public interface Streamable<T extends Event> {
+abstract class Streamable<T extends Event> {
/**
* Put an event into the streams.
*
* @param event
*/
- public void put(T event);
+ abstract void put(T event);
/**
* Stop and close all the streams.
*/
- public void close();
+ abstract void close();
/**
* @return the name of this streamable object.
*/
- public String getName();
+ abstract String getName();
+
+ /**
+ * Start all streams;
+ */
+ abstract void start();
}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/fb61d6df/subprojects/s4-core/src/test/java/org/apache/s4/core/CountTriggerTest.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/test/java/org/apache/s4/core/CountTriggerTest.java b/subprojects/s4-core/src/test/java/org/apache/s4/core/CountTriggerTest.java
new file mode 100644
index 0000000..fc34da3
--- /dev/null
+++ b/subprojects/s4-core/src/test/java/org/apache/s4/core/CountTriggerTest.java
@@ -0,0 +1,16 @@
+package org.apache.s4.core;
+
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class CountTriggerTest extends TriggerTest {
+
+ @Test
+ public void testEventCountBasedTrigger() throws Exception {
+ triggerType = TriggerType.COUNT_BASED;
+ Assert.assertTrue(createTriggerAppAndSendEvent().await(5, TimeUnit.SECONDS));
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/fb61d6df/subprojects/s4-core/src/test/java/org/apache/s4/core/NoTriggerTest.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/test/java/org/apache/s4/core/NoTriggerTest.java b/subprojects/s4-core/src/test/java/org/apache/s4/core/NoTriggerTest.java
new file mode 100644
index 0000000..a370777
--- /dev/null
+++ b/subprojects/s4-core/src/test/java/org/apache/s4/core/NoTriggerTest.java
@@ -0,0 +1,17 @@
+package org.apache.s4.core;
+
+import java.util.concurrent.TimeUnit;
+
+import junit.framework.Assert;
+
+import org.junit.Test;
+
+public class NoTriggerTest extends TriggerTest {
+
+ @Test
+ public void testNoTrigger() throws Exception {
+ triggerType = TriggerType.NONE;
+ Assert.assertFalse(createTriggerAppAndSendEvent().await(5, TimeUnit.SECONDS));
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/fb61d6df/subprojects/s4-core/src/test/java/org/apache/s4/core/TimeTriggerTest.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/test/java/org/apache/s4/core/TimeTriggerTest.java b/subprojects/s4-core/src/test/java/org/apache/s4/core/TimeTriggerTest.java
new file mode 100644
index 0000000..391e26a
--- /dev/null
+++ b/subprojects/s4-core/src/test/java/org/apache/s4/core/TimeTriggerTest.java
@@ -0,0 +1,19 @@
+package org.apache.s4.core;
+
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+
+public class TimeTriggerTest extends TriggerTest {
+
+ @Test
+ public void testTimeBasedTrigger() throws Exception {
+ triggerType = TriggerType.TIME_BASED;
+ Assert.assertTrue(createTriggerAppAndSendEvent().await(5, TimeUnit.SECONDS));
+
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/fb61d6df/subprojects/s4-core/src/test/java/org/apache/s4/core/TriggerTest.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/test/java/org/apache/s4/core/TriggerTest.java b/subprojects/s4-core/src/test/java/org/apache/s4/core/TriggerTest.java
new file mode 100644
index 0000000..6d55029
--- /dev/null
+++ b/subprojects/s4-core/src/test/java/org/apache/s4/core/TriggerTest.java
@@ -0,0 +1,75 @@
+package org.apache.s4.core;
+
+
+import java.io.IOException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import junit.framework.Assert;
+
+import org.apache.s4.fixtures.TestUtils;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.server.NIOServerCnxn.Factory;
+import org.junit.After;
+import org.junit.Before;
+
+
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+
+/**
+ * tests from subclasses are forked in separate VMs, an easy way to avoid
+ * conflict with unavailable resources when instantiating new S4 nodes
+ */
+
+public abstract class TriggerTest {
+
+ private Factory zookeeperServerConnectionFactory;
+ public static TriggerType triggerType;
+ protected TriggeredApp app;
+
+ public enum TriggerType {
+ TIME_BASED, COUNT_BASED, NONE
+ }
+
+ @Before
+ public void prepare() throws IOException, InterruptedException, KeeperException {
+ TestUtils.cleanupTmpDirs();
+ zookeeperServerConnectionFactory = TestUtils.startZookeeperServer();
+ }
+
+ @After
+ public void cleanup() throws IOException, InterruptedException {
+ if (app != null) {
+ app.close();
+ app = null;
+ }
+ TestUtils.stopZookeeperServer(zookeeperServerConnectionFactory);
+ }
+
+ protected CountDownLatch createTriggerAppAndSendEvent() throws IOException, KeeperException, InterruptedException {
+ final ZooKeeper zk = TestUtils.createZkClient();
+ Injector injector = Guice.createInjector(new TriggeredModule());
+ app = injector.getInstance(TriggeredApp.class);
+ app.init();
+ app.start();
+
+ String time1 = String.valueOf(System.currentTimeMillis());
+
+ CountDownLatch signalEvent1Processed = new CountDownLatch(1);
+ TestUtils.watchAndSignalCreation("/onEvent@" + time1, signalEvent1Processed, zk);
+
+ CountDownLatch signalEvent1Triggered = new CountDownLatch(1);
+ TestUtils.watchAndSignalCreation("/onTrigger[StringEvent]@" + time1, signalEvent1Triggered, zk);
+
+ TestUtils.injectIntoStringSocketAdapter(time1);
+
+ // check event processed
+ Assert.assertTrue(signalEvent1Processed.await(5, TimeUnit.SECONDS));
+
+ // return latch on trigger signal
+ return signalEvent1Triggered;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/fb61d6df/subprojects/s4-core/src/test/java/org/apache/s4/core/TriggerablePE.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/test/java/org/apache/s4/core/TriggerablePE.java b/subprojects/s4-core/src/test/java/org/apache/s4/core/TriggerablePE.java
new file mode 100644
index 0000000..155fcdb
--- /dev/null
+++ b/subprojects/s4-core/src/test/java/org/apache/s4/core/TriggerablePE.java
@@ -0,0 +1,80 @@
+package org.apache.s4.core;
+
+
+import java.io.IOException;
+
+import org.apache.s4.core.App;
+import org.apache.s4.core.ProcessingElement;
+import org.apache.s4.fixtures.TestUtils;
+import org.apache.s4.wordcount.StringEvent;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.ZooDefs.Ids;
+import org.apache.zookeeper.ZooKeeper;
+
+
+public class TriggerablePE extends ProcessingElement implements Watcher {
+
+ private ZooKeeper zk;
+
+ public TriggerablePE() {}
+
+ public TriggerablePE(App app) {
+ super(app);
+ }
+
+ public void onEvent(StringEvent event) {
+ try {
+ zk.create("/onEvent@"+event.getString(), new byte[0], Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+ } catch (KeeperException e) {
+ // TODO Auto-generated catch block
+ e.printStackTrace();
+ } catch (InterruptedException e) {
+ // TODO Auto-generated catch block
+ e.printStackTrace();
+ }
+ }
+
+ @Override
+ protected void onCreate() {
+ if (zk == null) {
+ try {
+ zk = new ZooKeeper("localhost:" + TestUtils.ZK_PORT, 4000, this);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ public void onTrigger(StringEvent event) {
+ try {
+ zk.create("/onTrigger[StringEvent]@"+event.getString(), new byte[0], Ids.OPEN_ACL_UNSAFE,
+ CreateMode.PERSISTENT);
+ } catch (KeeperException e) {
+ // TODO Auto-generated catch block
+ e.printStackTrace();
+ } catch (InterruptedException e) {
+ // TODO Auto-generated catch block
+ e.printStackTrace();
+ }
+ }
+
+
+ @Override
+ protected void onRemove() {
+ // TODO Auto-generated method stub
+
+ }
+
+ @Override
+ public void process(WatchedEvent event) {
+ // TODO Auto-generated method stub
+
+ }
+
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/fb61d6df/subprojects/s4-core/src/test/java/org/apache/s4/core/TriggeredApp.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/test/java/org/apache/s4/core/TriggeredApp.java b/subprojects/s4-core/src/test/java/org/apache/s4/core/TriggeredApp.java
new file mode 100644
index 0000000..80207f9
--- /dev/null
+++ b/subprojects/s4-core/src/test/java/org/apache/s4/core/TriggeredApp.java
@@ -0,0 +1,54 @@
+package org.apache.s4.core;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.s4.base.Event;
+import org.apache.s4.fixtures.SocketAdapter;
+import org.apache.s4.wordcount.SentenceKeyFinder;
+import org.apache.s4.wordcount.StringEvent;
+
+import com.google.inject.Inject;
+
+public class TriggeredApp extends App {
+
+ SocketAdapter<StringEvent> socketAdapter;
+
+ @Inject
+ public TriggeredApp() {
+ super();
+ }
+
+ @Override
+ protected void onStart() {
+ // TODO Auto-generated method stub
+
+ }
+
+ @Override
+ protected void onInit() {
+
+ TriggerablePE prototype = createPE(TriggerablePE.class);
+ Stream<StringEvent> stream = createStream("stream", new SentenceKeyFinder(), prototype);
+ switch (TriggerTest.triggerType) {
+ case COUNT_BASED:
+ prototype.withTrigger(Event.class, 1, 0, TimeUnit.SECONDS);
+ break;
+ case TIME_BASED:
+ prototype.withTrigger(Event.class, 1, 1, TimeUnit.MILLISECONDS);
+ default:
+ break;
+ }
+
+ try {
+ socketAdapter = new SocketAdapter<StringEvent>(stream, new SocketAdapter.SentenceEventFactory());
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ protected void onClose() {
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/fb61d6df/subprojects/s4-core/src/test/java/org/apache/s4/core/TriggeredModule.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/test/java/org/apache/s4/core/TriggeredModule.java b/subprojects/s4-core/src/test/java/org/apache/s4/core/TriggeredModule.java
new file mode 100644
index 0000000..cf61b24
--- /dev/null
+++ b/subprojects/s4-core/src/test/java/org/apache/s4/core/TriggeredModule.java
@@ -0,0 +1,6 @@
+package org.apache.s4.core;
+
+import org.apache.s4.fixtures.FileBasedClusterManagementTestModule;
+
+public class TriggeredModule extends FileBasedClusterManagementTestModule<TriggeredApp> {
+}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/fb61d6df/subprojects/s4-core/src/test/java/org/apache/s4/core/apploading/SimpleApp.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/test/java/org/apache/s4/core/apploading/SimpleApp.java b/subprojects/s4-core/src/test/java/org/apache/s4/core/apploading/SimpleApp.java
index 92aae16..08831cb 100644
--- a/subprojects/s4-core/src/test/java/org/apache/s4/core/apploading/SimpleApp.java
+++ b/subprojects/s4-core/src/test/java/org/apache/s4/core/apploading/SimpleApp.java
@@ -19,7 +19,7 @@ public class SimpleApp extends App {
}
@Override
- protected void start() {
+ protected void onStart() {
try {
final ZooKeeper zk = TestUtils.createZkClient();
zk.create("/simpleAppCreated", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
@@ -30,7 +30,7 @@ public class SimpleApp extends App {
}
@Override
- protected void init() {
+ protected void onInit() {
SimplePE prototype = createPE(SimplePE.class);
Stream<StringEvent> stream = createStream("stream", new SentenceKeyFinder(), prototype);
try {
@@ -41,7 +41,7 @@ public class SimpleApp extends App {
}
@Override
- protected void close() {
+ protected void onClose() {
// TODO Auto-generated method stub
}
}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/fb61d6df/subprojects/s4-core/src/test/java/org/apache/s4/core/triggers/CountTriggerTest.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/test/java/org/apache/s4/core/triggers/CountTriggerTest.java b/subprojects/s4-core/src/test/java/org/apache/s4/core/triggers/CountTriggerTest.java
deleted file mode 100644
index f8b0e40..0000000
--- a/subprojects/s4-core/src/test/java/org/apache/s4/core/triggers/CountTriggerTest.java
+++ /dev/null
@@ -1,16 +0,0 @@
-package org.apache.s4.core.triggers;
-
-import java.util.concurrent.TimeUnit;
-
-import org.junit.Assert;
-import org.junit.Test;
-
-public class CountTriggerTest extends TriggerTest {
-
- @Test
- public void testEventCountBasedTrigger() throws Exception {
- triggerType = TriggerType.COUNT_BASED;
- Assert.assertTrue(createTriggerAppAndSendEvent().await(5, TimeUnit.SECONDS));
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/fb61d6df/subprojects/s4-core/src/test/java/org/apache/s4/core/triggers/NoTriggerTest.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/test/java/org/apache/s4/core/triggers/NoTriggerTest.java b/subprojects/s4-core/src/test/java/org/apache/s4/core/triggers/NoTriggerTest.java
deleted file mode 100644
index 6feb783..0000000
--- a/subprojects/s4-core/src/test/java/org/apache/s4/core/triggers/NoTriggerTest.java
+++ /dev/null
@@ -1,17 +0,0 @@
-package org.apache.s4.core.triggers;
-
-import java.util.concurrent.TimeUnit;
-
-import junit.framework.Assert;
-
-import org.junit.Test;
-
-public class NoTriggerTest extends TriggerTest {
-
- @Test
- public void testNoTrigger() throws Exception {
- triggerType = TriggerType.NONE;
- Assert.assertFalse(createTriggerAppAndSendEvent().await(5, TimeUnit.SECONDS));
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/fb61d6df/subprojects/s4-core/src/test/java/org/apache/s4/core/triggers/TimeTriggerTest.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/test/java/org/apache/s4/core/triggers/TimeTriggerTest.java b/subprojects/s4-core/src/test/java/org/apache/s4/core/triggers/TimeTriggerTest.java
deleted file mode 100644
index 8df087c..0000000
--- a/subprojects/s4-core/src/test/java/org/apache/s4/core/triggers/TimeTriggerTest.java
+++ /dev/null
@@ -1,19 +0,0 @@
-package org.apache.s4.core.triggers;
-
-import java.util.concurrent.TimeUnit;
-
-import org.junit.Assert;
-import org.junit.Test;
-
-
-public class TimeTriggerTest extends TriggerTest {
-
- @Test
- public void testTimeBasedTrigger() throws Exception {
- triggerType = TriggerType.TIME_BASED;
- Assert.assertTrue(createTriggerAppAndSendEvent().await(5, TimeUnit.SECONDS));
-
- }
-
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/fb61d6df/subprojects/s4-core/src/test/java/org/apache/s4/core/triggers/TriggerTest.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/test/java/org/apache/s4/core/triggers/TriggerTest.java b/subprojects/s4-core/src/test/java/org/apache/s4/core/triggers/TriggerTest.java
deleted file mode 100644
index adb7bcf..0000000
--- a/subprojects/s4-core/src/test/java/org/apache/s4/core/triggers/TriggerTest.java
+++ /dev/null
@@ -1,75 +0,0 @@
-package org.apache.s4.core.triggers;
-
-
-import java.io.IOException;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-
-import junit.framework.Assert;
-
-import org.apache.s4.fixtures.TestUtils;
-import org.apache.zookeeper.KeeperException;
-import org.apache.zookeeper.ZooKeeper;
-import org.apache.zookeeper.server.NIOServerCnxn.Factory;
-import org.junit.After;
-import org.junit.Before;
-
-
-import com.google.inject.Guice;
-import com.google.inject.Injector;
-
-/**
- * tests from subclasses are forked in separate VMs, an easy way to avoid
- * conflict with unavailable resources when instantiating new S4 nodes
- */
-
-public abstract class TriggerTest {
-
- private Factory zookeeperServerConnectionFactory;
- public static TriggerType triggerType;
- protected TriggeredApp app;
-
- public enum TriggerType {
- TIME_BASED, COUNT_BASED, NONE
- }
-
- @Before
- public void prepare() throws IOException, InterruptedException, KeeperException {
- TestUtils.cleanupTmpDirs();
- zookeeperServerConnectionFactory = TestUtils.startZookeeperServer();
- }
-
- @After
- public void cleanup() throws IOException, InterruptedException {
- if (app != null) {
- app.close();
- app = null;
- }
- TestUtils.stopZookeeperServer(zookeeperServerConnectionFactory);
- }
-
- protected CountDownLatch createTriggerAppAndSendEvent() throws IOException, KeeperException, InterruptedException {
- final ZooKeeper zk = TestUtils.createZkClient();
- Injector injector = Guice.createInjector(new TriggeredModule());
- app = injector.getInstance(TriggeredApp.class);
- app.init();
- app.start();
-
- String time1 = String.valueOf(System.currentTimeMillis());
-
- CountDownLatch signalEvent1Processed = new CountDownLatch(1);
- TestUtils.watchAndSignalCreation("/onEvent@" + time1, signalEvent1Processed, zk);
-
- CountDownLatch signalEvent1Triggered = new CountDownLatch(1);
- TestUtils.watchAndSignalCreation("/onTrigger[StringEvent]@" + time1, signalEvent1Triggered, zk);
-
- TestUtils.injectIntoStringSocketAdapter(time1);
-
- // check event processed
- Assert.assertTrue(signalEvent1Processed.await(5, TimeUnit.SECONDS));
-
- // return latch on trigger signal
- return signalEvent1Triggered;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/fb61d6df/subprojects/s4-core/src/test/java/org/apache/s4/core/triggers/TriggerablePE.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/test/java/org/apache/s4/core/triggers/TriggerablePE.java b/subprojects/s4-core/src/test/java/org/apache/s4/core/triggers/TriggerablePE.java
deleted file mode 100644
index 863f80e..0000000
--- a/subprojects/s4-core/src/test/java/org/apache/s4/core/triggers/TriggerablePE.java
+++ /dev/null
@@ -1,80 +0,0 @@
-package org.apache.s4.core.triggers;
-
-
-import java.io.IOException;
-
-import org.apache.s4.core.App;
-import org.apache.s4.core.ProcessingElement;
-import org.apache.s4.fixtures.TestUtils;
-import org.apache.s4.wordcount.StringEvent;
-import org.apache.zookeeper.CreateMode;
-import org.apache.zookeeper.KeeperException;
-import org.apache.zookeeper.WatchedEvent;
-import org.apache.zookeeper.Watcher;
-import org.apache.zookeeper.ZooDefs.Ids;
-import org.apache.zookeeper.ZooKeeper;
-
-
-public class TriggerablePE extends ProcessingElement implements Watcher {
-
- private ZooKeeper zk;
-
- public TriggerablePE() {}
-
- public TriggerablePE(App app) {
- super(app);
- }
-
- public void onEvent(StringEvent event) {
- try {
- zk.create("/onEvent@"+event.getString(), new byte[0], Ids.OPEN_ACL_UNSAFE,
- CreateMode.PERSISTENT);
- } catch (KeeperException e) {
- // TODO Auto-generated catch block
- e.printStackTrace();
- } catch (InterruptedException e) {
- // TODO Auto-generated catch block
- e.printStackTrace();
- }
- }
-
- @Override
- protected void onCreate() {
- if (zk == null) {
- try {
- zk = new ZooKeeper("localhost:" + TestUtils.ZK_PORT, 4000, this);
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
- }
-
- public void onTrigger(StringEvent event) {
- try {
- zk.create("/onTrigger[StringEvent]@"+event.getString(), new byte[0], Ids.OPEN_ACL_UNSAFE,
- CreateMode.PERSISTENT);
- } catch (KeeperException e) {
- // TODO Auto-generated catch block
- e.printStackTrace();
- } catch (InterruptedException e) {
- // TODO Auto-generated catch block
- e.printStackTrace();
- }
- }
-
-
- @Override
- protected void onRemove() {
- // TODO Auto-generated method stub
-
- }
-
- @Override
- public void process(WatchedEvent event) {
- // TODO Auto-generated method stub
-
- }
-
-
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/fb61d6df/subprojects/s4-core/src/test/java/org/apache/s4/core/triggers/TriggeredApp.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/test/java/org/apache/s4/core/triggers/TriggeredApp.java b/subprojects/s4-core/src/test/java/org/apache/s4/core/triggers/TriggeredApp.java
deleted file mode 100644
index 21c7ee2..0000000
--- a/subprojects/s4-core/src/test/java/org/apache/s4/core/triggers/TriggeredApp.java
+++ /dev/null
@@ -1,58 +0,0 @@
-package org.apache.s4.core.triggers;
-
-
-import java.io.IOException;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.s4.base.Event;
-import org.apache.s4.core.App;
-import org.apache.s4.core.Stream;
-import org.apache.s4.fixtures.SocketAdapter;
-import org.apache.s4.wordcount.SentenceKeyFinder;
-import org.apache.s4.wordcount.StringEvent;
-
-
-import com.google.inject.Inject;
-
-public class TriggeredApp extends App {
-
- SocketAdapter<StringEvent> socketAdapter;
-
- @Inject
- public TriggeredApp() {
- super();
- }
-
- @Override
- protected void start() {
- // TODO Auto-generated method stub
-
- }
-
- @Override
- protected void init() {
-
- TriggerablePE prototype = createPE(TriggerablePE.class);
- Stream<StringEvent> stream = createStream("stream", new SentenceKeyFinder(), prototype);
- switch (TriggerTest.triggerType) {
- case COUNT_BASED:
- prototype.setTrigger(Event.class, 1, 0, TimeUnit.SECONDS);
- break;
- case TIME_BASED:
- prototype.setTrigger(Event.class, 1, 1, TimeUnit.MILLISECONDS);
- default:
- break;
- }
-
- try {
- socketAdapter = new SocketAdapter<StringEvent>(stream, new SocketAdapter.SentenceEventFactory());
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- protected void close() {
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/fb61d6df/subprojects/s4-core/src/test/java/org/apache/s4/core/triggers/TriggeredModule.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/test/java/org/apache/s4/core/triggers/TriggeredModule.java b/subprojects/s4-core/src/test/java/org/apache/s4/core/triggers/TriggeredModule.java
deleted file mode 100644
index cf7cd80..0000000
--- a/subprojects/s4-core/src/test/java/org/apache/s4/core/triggers/TriggeredModule.java
+++ /dev/null
@@ -1,6 +0,0 @@
-package org.apache.s4.core.triggers;
-
-import org.apache.s4.fixtures.FileBasedClusterManagementTestModule;
-
-public class TriggeredModule extends FileBasedClusterManagementTestModule<TriggeredApp> {
-}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/fb61d6df/subprojects/s4-core/src/test/java/org/apache/s4/wordcount/WordCountApp.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/test/java/org/apache/s4/wordcount/WordCountApp.java b/subprojects/s4-core/src/test/java/org/apache/s4/wordcount/WordCountApp.java
index 4d0d31a..278b752 100644
--- a/subprojects/s4-core/src/test/java/org/apache/s4/wordcount/WordCountApp.java
+++ b/subprojects/s4-core/src/test/java/org/apache/s4/wordcount/WordCountApp.java
@@ -1,13 +1,11 @@
package org.apache.s4.wordcount;
-
import java.io.IOException;
import org.apache.s4.core.App;
import org.apache.s4.core.Stream;
import org.apache.s4.fixtures.SocketAdapter;
-
import com.google.inject.Inject;
public class WordCountApp extends App {
@@ -21,19 +19,19 @@ public class WordCountApp extends App {
}
@Override
- protected void start() {
+ protected void onStart() {
// TODO Auto-generated method stub
}
@Override
- protected void init() {
+ protected void onInit() {
WordClassifierPE wordClassifierPrototype = createPE(WordClassifierPE.class);
Stream<WordCountEvent> wordCountStream = createStream("words counts stream", new WordCountKeyFinder(),
wordClassifierPrototype);
WordCounterPE wordCounterPrototype = createPE(WordCounterPE.class);
-// wordCounterPrototype.setTrigger(WordSeenEvent.class, 1, 0, null);
+ // wordCounterPrototype.setTrigger(WordSeenEvent.class, 1, 0, null);
wordCounterPrototype.setWordClassifierStream(wordCountStream);
Stream<WordSeenEvent> wordSeenStream = createStream("words seen stream", new WordSeenKeyFinder(),
wordCounterPrototype);
@@ -63,7 +61,7 @@ public class WordCountApp extends App {
}
@Override
- protected void close() {
+ protected void onClose() {
socketAdapter.close();
}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/fb61d6df/subprojects/s4-example/src/main/java/org/apache/s4/example/counter/CountKeyFinder.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-example/src/main/java/org/apache/s4/example/counter/CountKeyFinder.java b/subprojects/s4-example/src/main/java/org/apache/s4/example/counter/CountKeyFinder.java
index e8183e5..5485c89 100644
--- a/subprojects/s4-example/src/main/java/org/apache/s4/example/counter/CountKeyFinder.java
+++ b/subprojects/s4-example/src/main/java/org/apache/s4/example/counter/CountKeyFinder.java
@@ -20,7 +20,6 @@ import java.util.List;
import org.apache.s4.core.KeyFinder;
-
public class CountKeyFinder implements KeyFinder<CountEvent> {
public List<String> get(CountEvent event) {
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/fb61d6df/subprojects/s4-example/src/main/java/org/apache/s4/example/counter/MyApp.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-example/src/main/java/org/apache/s4/example/counter/MyApp.java b/subprojects/s4-example/src/main/java/org/apache/s4/example/counter/MyApp.java
index 7ac1d7a..c8586a4 100644
--- a/subprojects/s4-example/src/main/java/org/apache/s4/example/counter/MyApp.java
+++ b/subprojects/s4-example/src/main/java/org/apache/s4/example/counter/MyApp.java
@@ -26,7 +26,6 @@ import org.apache.s4.core.Stream;
import com.google.inject.Guice;
import com.google.inject.Injector;
-
/*
* This is a sample application to test a new S4 API.
* See README file for details.
@@ -41,58 +40,58 @@ final public class MyApp extends App {
/*
*
*
- * The application graph itself is created in this Class. However,
- * developers may provide tools for creating apps which will generate the
- * objects.
+ * The application graph itself is created in this Class. However, developers may provide tools for creating apps
+ * which will generate the objects.
*
- * IMPORTANT: we create a graph of PE prototypes. The prototype is a class
- * instance that is used as a prototype from which all PE instance will be
- * created. The prototype itself is not used as an instance. (Except when
- * the PE is of type Singleton PE). To create a data structure for each PE
- * instance you must do it in the method ProcessingElement.onCreate().
+ * IMPORTANT: we create a graph of PE prototypes. The prototype is a class instance that is used as a prototype from
+ * which all PE instance will be created. The prototype itself is not used as an instance. (Except when the PE is of
+ * type Singleton PE). To create a data structure for each PE instance you must do it in the method
+ * ProcessingElement.onCreate().
*/
-
/*
- * Build the application graph using POJOs. Don't like it? Write a nice
- * tool.
+ * Build the application graph using POJOs. Don't like it? Write a nice tool.
*
* @see io.s4.App#init()
*/
@SuppressWarnings("unchecked")
@Override
- protected void init() {
+ protected void onInit() {
/* PE that prints counts to console. */
PrintPE printPE = createPE(PrintPE.class);
- Stream<CountEvent> userCountStream = createStream("User Count Stream",
- new CountKeyFinder(), printPE);
- Stream<CountEvent> genderCountStream = createStream(
- "Gender Count Stream", new CountKeyFinder(), printPE);
- Stream<CountEvent> ageCountStream = createStream("Age Count Stream",
- new CountKeyFinder(), printPE);
+ Stream<CountEvent> userCountStream = createStream(CountEvent.class).withName("User Count Stream")
+ .withKey(new CountKeyFinder()).to(printPE);
+
+ Stream<CountEvent> genderCountStream = createStream(CountEvent.class).withName("Gender Count Stream")
+ .withKey(new CountKeyFinder()).to(printPE);
+
+ Stream<CountEvent> ageCountStream = createStream(CountEvent.class).withName("Age Count Stream")
+ .withKey(new CountKeyFinder()).to(printPE);
/* PEs that count events by user, gender, and age. */
- CounterPE userCountPE = createPE(CounterPE.class);
- userCountPE.setTrigger(Event.class, interval, 10l, TimeUnit.SECONDS);
+ CounterPE userCountPE = createPE(CounterPE.class);// .withTrigger(Event.class, interval, 10l, TimeUnit.SECONDS);
+ userCountPE.withTrigger(Event.class, interval, 10l, TimeUnit.SECONDS);
userCountPE.setCountStream(userCountStream);
CounterPE genderCountPE = createPE(CounterPE.class);
- genderCountPE.setTrigger(Event.class, interval, 10l, TimeUnit.SECONDS);
+ genderCountPE.withTrigger(Event.class, interval, 10l, TimeUnit.SECONDS);
genderCountPE.setCountStream(genderCountStream);
CounterPE ageCountPE = createPE(CounterPE.class);
- ageCountPE.setTrigger(Event.class, interval, 10l, TimeUnit.SECONDS);
+ ageCountPE.withTrigger(Event.class, interval, 10l, TimeUnit.SECONDS);
ageCountPE.setCountStream(ageCountStream);
/* Streams that output user events keyed on user, gender, and age. */
- Stream<UserEvent> userStream = createStream("User Stream",
- new UserIDKeyFinder(), userCountPE);
- Stream<UserEvent> genderStream = createStream("Gender Stream",
- new GenderKeyFinder(), genderCountPE);
- Stream<UserEvent> ageStream = createStream("Age Stream",
- new AgeKeyFinder(), ageCountPE);
+ Stream<UserEvent> userStream = createStream(UserEvent.class).withName("User Stream")
+ .withKey(new UserIDKeyFinder()).to(userCountPE);
+
+ Stream<UserEvent> genderStream = createStream(UserEvent.class).withName("Gender Stream")
+ .withKey(new GenderKeyFinder()).to(genderCountPE);
+
+ Stream<UserEvent> ageStream = createStream(UserEvent.class).withName("Age Stream").withKey(new AgeKeyFinder())
+ .to(ageCountPE);
generateUserEventPE = createPE(GenerateUserEventPE.class);
generateUserEventPE.setStreams(userStream, genderStream, ageStream);
@@ -104,7 +103,7 @@ final public class MyApp extends App {
* @see io.s4.App#start()
*/
@Override
- protected void start() {
+ protected void onStart() {
for (int i = 0; i < 200; i++) {
generateUserEventPE.onTrigger(null);
@@ -117,13 +116,12 @@ final public class MyApp extends App {
e.printStackTrace();
}
- System.out.println("Done. Closing...");
- removeAll();
-
+ System.out.println("Done. Wait until the main app closes.");
+ // close();
}
@Override
- protected void close() {
+ protected void onClose() {
System.out.println("Bye.");
}
@@ -137,5 +135,13 @@ final public class MyApp extends App {
myApp.setCommLayer(sender, receiver);
myApp.init();
myApp.start();
+
+ try {
+ Thread.sleep(10000);
+ } catch (InterruptedException e) {
+ // TODO Auto-generated catch block
+ e.printStackTrace();
+ }
+ myApp.close();
}
}