Skip to content
This repository
Browse code

Merge remote-tracking branch 'upstream/master'

  • Loading branch information...
commit 8d3746027c9ddd4c2c0e8adb7ccd91c1053424bd 2 parents 0e61639 + 7aaba95
authored July 15, 2011

Showing 67 changed files with 1,187 additions and 266 deletions. Show diff stats Hide diff stats

  1. 8  .gitignore
  2. 40  README.md
  3. 12  build.gradle
  4. 58  dev-notes.txt
  5. BIN  gradle/wrapper/gradle-wrapper.jar
  6. 4  gradle/wrapper/gradle-wrapper.properties
  7. 31  s4-core/README.md
  8. 33  s4-core/src/main/java/io/s4/MainApp.java
  9. 18  s4-core/src/main/java/io/s4/dispatcher/partitioner/RoundRobinPartitioner.java
  10. 2  s4-core/src/main/java/io/s4/ft/SafeKeeperId.java
  11. 4  s4-core/src/main/java/io/s4/message/SinglePERequest.java
  12. 49  s4-core/src/main/java/io/s4/processor/AbstractPE.java
  13. 2  s4-core/src/main/java/io/s4/processor/ControlEventProcessor.java
  14. 9  s4-core/src/main/java/io/s4/processor/JoinPE.java
  15. 22  s4-core/src/main/java/io/s4/processor/PEContainer.java
  16. 10  s4-core/src/main/java/io/s4/processor/PrintEventPE.java
  17. 24  s4-core/src/main/java/io/s4/processor/PrototypeWrapper.java
  18. 9  s4-core/src/main/java/io/s4/processor/ReroutePE.java
  19. 9  s4-core/src/main/java/io/s4/processor/SimpleCountingPE.java
  20. 0  s4-core/src/main/resources/s4-apps/{README → .gitignore}
  21. 2  s4-core/src/main/resources/s4-core/conf/default/client-adapter-conf.xml
  22. 2  s4-core/src/main/resources/s4-core/conf/default/client-stub-conf.xml
  23. 6  s4-core/src/main/resources/s4-core/conf/default/clusters.xml
  24. 26  s4-core/src/main/resources/s4-core/conf/default/log4j.xml
  25. 2  s4-core/src/main/resources/s4-core/conf/default/s4-core-conf.xml
  26. 41  s4-core/src/main/resources/s4-core/conf/dynamic/adapter-conf.xml
  27. 64  s4-core/src/main/resources/s4-core/conf/dynamic/client-adapter-conf.xml
  28. 13  s4-core/src/main/resources/s4-core/conf/dynamic/client-stub-conf.xml
  29. 18  s4-core/src/main/resources/s4-core/conf/dynamic/clusters.xml
  30. 6  s4-core/src/main/resources/s4-core/conf/dynamic/event-clock.xml
  31. 48  s4-core/src/main/resources/s4-core/conf/dynamic/log4j.xml
  32. 152  s4-core/src/main/resources/s4-core/conf/dynamic/s4-core-conf.xml
  33. 9  s4-core/src/main/resources/s4-core/conf/dynamic/s4-core.properties-header
  34. 6  s4-core/src/main/resources/s4-core/conf/dynamic/wall-clock.xml
  35. 0  s4-core/src/main/resources/s4-core/lock/{README → .gitignore}
  36. 0  s4-core/src/main/resources/s4-core/logs/{README → .gitignore}
  37. 0  s4-core/src/main/resources/s4-exts/{README → .gitignore}
  38. 2  s4-core/src/main/resources/scripts/generate-load.sh
  39. 2  s4-core/src/main/resources/scripts/run-adapter.sh
  40. 14  s4-core/src/main/resources/scripts/run-client-adapter.sh
  41. 0  s4-core/src/main/resources/scripts/{s4-start.sh → start-s4.sh}
  42. 42  s4-core/src/main/resources/scripts/task-setup.sh
  43. 18  s4-core/src/test/java/io/s4/ft/S4App.java
  44. 4  s4-core/src/test/java/io/s4/ft/S4TestCase.java
  45. 2  s4-core/src/test/java/io/s4/ft/s4_core_conf_bk_backend.xml
  46. 2  s4-core/src/test/java/io/s4/ft/s4_core_conf_fs_backend.xml
  47. 2  s4-core/src/test/java/io/s4/ft/wordcount/s4_core_conf_bk_backend.xml
  48. 2  s4-core/src/test/java/io/s4/ft/wordcount/s4_core_conf_fs_backend.xml
  49. 2  s4-core/src/test/java/io/s4/ft/wordcount/s4_core_conf_redis_backend.xml
  50. 7  s4-core/src/test/java/io/s4/processor/MockPE.java
  51. 1  s4-core/src/test/java/io/s4/processor/TestPrototypeWrapper.java
  52. 2  s4-core/src/test/java/io/s4/wordcount/s4_core_conf.xml
  53. 2  s4-examples/speech01-scala/src/main/scala/SentenceReceiverPE.scala
  54. 6  s4-examples/speech01/src/main/java/io/s4/example/speech01/SentenceReceiverPE.java
  55. 0  s4-examples/testinput/{ → src/main/resources}/pe-query
  56. 0  s4-examples/testinput/{ → src/main/resources}/proto-query
  57. 0  s4-examples/testinput/{ → src/main/resources}/sentence.in
  58. 0  s4-examples/testinput/{ → src/main/resources}/speech.in
  59. 0  s4-examples/testinput/{ → src/main/resources}/speeches.txt
  60. 1  s4-examples/twittertopiccount-scala/src/main/scala/processor/TopNTopicPE.scala
  61. 1  s4-examples/twittertopiccount-scala/src/main/scala/processor/TopicCountAndReportPE.scala
  62. 1  s4-examples/twittertopiccount-scala/src/main/scala/processor/TopicExtractorPE.scala
  63. 24  s4-core/src/main/java/io/s4/processor/ProcessingElement.java → s4-tools/loadgenerator/build.gradle
  64. 349  s4-tools/loadgenerator/src/main/java/io/s4/tools/loadgenerator/LoadGenerator.java
  65. 131  s4-tools/loadgenerator/src/main/java/io/s4/tools/loadgenerator/Pacer.java
  66. 93  s4-tools/loadgenerator/src/main/resources/scripts/generate-load.sh
  67. 4  settings.gradle
8  .gitignore
@@ -6,6 +6,7 @@
6 6
 # Build directory
7 7
 target/
8 8
 build/
  9
+bin/
9 10
 
10 11
 # SVN
11 12
 .svn
@@ -21,3 +22,10 @@ build/
21 22
 
22 23
 # class files
23 24
 *.class 
  25
+.DS_Store 
  26
+
  27
+# create an empty .gitignore file when you want to track an empty
  28
+# directory but want to ignore its content. Example: to keep the 
  29
+# "logs" dir do: "touch logs/.gitignore"
  30
+#
  31
+!.gitignore
40  README.md
Source Rendered
@@ -65,23 +65,43 @@ git clone https://github.com/s4/s4.git
65 65
 # Create image
66 66
 gradlew allImage
67 67
 
68  
-# Change permissions
69  
-chmod u+x ./build/s4-image/scripts/*
  68
+# set the S4_IMAGE environmental variable
  69
+cd build/s4-image/
  70
+export S4_IMAGE=`pwd`
70 71
 
71  
-# Copy S4 application to deployment dir (s4-apps)
72  
-cp -rp build/s4-image/s4-example-apps/s4-example-twittertopiccount build/s4-image/s4-apps/
  72
+# get the sample application
  73
+git clone git://github.com/s4/twittertopiccount.git
73 74
 
74  
-# Enter your twitter user/pass in config file
75  
-$EDITOR build/s4-image/s4-apps/s4-example-twittertopiccount/adapter-conf.xml 
  75
+# build the sample application
  76
+./gradlew install
76 77
 
77  
-# Start server with s4-example-twittertopiccount app
78  
-./build/s4-image/scripts/s4-start.sh &
  78
+# deploy the sample application into the S4 image (relies in the S4_IMAGE environmental variable)
  79
+./gradlew deploy
79 80
 
80  
-# Start adapter
81  
- ./build/s4-image/scripts/run-adapter.sh -x -u build/s4-image/s4-apps/s4-example-twittertopiccount/lib/s4-example-twittertopiccount-0.3-SNAPSHOT.jar -d build/s4-image/s4-apps/s4-example-twittertopiccount/adapter-conf.xml &
  81
+# set the TWIT_LISTENER environmental variable
  82
+cd build/install/twitter_feed_listener
  83
+export TWIT_LISTENER=`pwd`
  84
+
  85
+# Start server with twittertopiccount app
  86
+$S4_IMAGE/scripts/start-s4.sh -r client-adapter &
  87
+
  88
+# start the client adapter
  89
+$S4_IMAGE/scripts/run-client-adapter.sh -s client-adapter -g s4 -d $S4_IMAGE/s4-core/conf/default/client-stub-conf.xml &
  90
+
  91
+# run a client to send events into the S4 cluster. Replace <your-twitter-user> and <your-twitter-password> with your Twitter userid and password.
  92
+$TWIT_LISTENER/bin/twitter_feed_listener <your-twitter-user> <your-twitter-password> &
82 93
 
83 94
 # Check output
84 95
 cat /tmp/top_n_hashtags
85 96
 </pre>
86 97
 
  98
+Developing with Eclipse
  99
+-----------------------
  100
+
  101
+The command `gradle eclipse` will create an eclipse project that you can import from the Eclipse IDE.
  102
+
  103
+There is now a [Gradle plugin for the Eclipse IDE](http://static.springsource.org/sts/docs/2.7.0.M1/reference/html/gradle/index.html). 
  104
+To install Gradle without installing the full Spring development environment follow the
  105
+[instructions](http://static.springsource.org/sts/docs/2.7.0.M1/reference/html/gradle/installation.html) under the heading 
  106
+"Installing from update site". There is also a discussion in the [Gradle mailing list](http://gradle.1045684.n5.nabble.com/ANN-Gradle-Eclipse-Plugin-td4387658.html).
87 107
 
12  build.gradle
@@ -69,7 +69,6 @@ libraries = [
69 69
 json:               'org.json:json:20090211',
70 70
 lift_json:          'net.liftweb:lift-json_2.8.1:2.2',
71 71
 gson:               'com.google.code.gson:gson:1.6',
72  
-junit:              'junit:junit:4.4',
73 72
 zk:                 'org.apache.zookeeper:zookeeper:3.3.1',
74 73
 log4j:              'log4j:log4j:1.2.15',
75 74
 flexjson:           'net.sf.flexjson:flexjson:2.1',
@@ -85,6 +84,7 @@ commons_jexl:       'commons-jexl:commons-jexl:1.1',
85 84
 commons_codec:      'commons-codec:commons-codec:1.4',
86 85
 commons_httpclient: 'commons-httpclient:commons-httpclient:3.1',
87 86
 spring:             'org.springframework:spring:2.5.6',
  87
+junit:              'junit:junit:4.4',
88 88
 scala_compiler:     'org.scala-lang:scala-compiler:2.8.1',
89 89
 scala_library:      'org.scala-lang:scala-library:2.8.1',
90 90
 jedis:              'redis.clients:jedis:1.5.2',
@@ -104,6 +104,11 @@ subprojects {
104 104
     //defaultTasks 'build'
105 105
 
106 106
     group = 'io.s4'
  107
+    
  108
+    /* Remove this once this bug is fixed: http://issues.gradle.org/browse/GRADLE-1157 */
  109
+    eclipseClasspath {
  110
+       downloadSources = false; // required for eclipseClasspath to work
  111
+	}
107 112
 
108 113
     /* Common dependencies applied to all subprojects. */
109 114
     dependencies { 
@@ -220,6 +225,7 @@ task allImage(type: Copy, dependsOn: s4Javadoc) {
220 225
     with allDistImage
221 226
 }
222 227
 
  228
+
223 229
 task binTgz( type: Tar) {
224 230
     description = "Build binary bundle in GZIP format"
225 231
     classifier = 'bin'
@@ -239,9 +245,9 @@ task allTgz( type: Tar, dependsOn: s4Javadoc) {
239 245
 }
240 246
 
241 247
 /* Generates the gradlew scripts.
242  
-http://www.gradle.org/1.0-milestone-1/docs/userguide/gradle_wrapper.html */ 
  248
+http://www.gradle.org/1.0-milestone-3/docs/userguide/gradle_wrapper.html */ 
243 249
 task wrapper(type: Wrapper) { 
244  
-    gradleVersion = '1.0-milestone-1' 
  250
+    gradleVersion = '1.0-milestone-3' 
245 251
 }
246 252
 
247 253
 class Version {
58  dev-notes.txt
@@ -14,30 +14,30 @@ cd s4
14 14
 gradlew clean allImage
15 15
 
16 16
 # Create some environment variables. 
17  
-export IMAGE_BASE=`pwd`'/build/s4-image'
18  
-export PYTHONPATH=${IMAGE_BASE}/s4-driver/lib/python
19  
-export PERLLIB=${IMAGE_BASE}/s4-driver/lib/perl
  17
+export S4_IMAGE=`pwd`'/build/s4-image'
  18
+export PYTHONPATH=${S4_IMAGE}/s4-driver/lib/python
  19
+export PERLLIB=${S4_IMAGE}/s4-driver/lib/perl
20 20
 
21 21
 # Change script permissions.
22  
-chmod u+x $IMAGE_BASE/scripts/*
23  
-chmod u+x $IMAGE_BASE/s4-driver/scripts/*
  22
+chmod u+x $S4_IMAGE/scripts/*
  23
+chmod u+x $S4_IMAGE/s4-driver/scripts/*
24 24
 
25 25
 # Copy speech02 app to apps deployment directory.
26  
-cp -fr $IMAGE_BASE/s4-example-apps/s4-example-speech02 $IMAGE_BASE/s4-apps
  26
+cp -fr $S4_IMAGE/s4-example-apps/s4-example-speech02 $S4_IMAGE/s4-apps
27 27
 
28 28
 # Start S4 server in standalone mode.
29  
-$IMAGE_BASE/scripts/s4-start.sh -r client-adapter &
  29
+$S4_IMAGE/scripts/start-s4.sh -r client-adapter &
30 30
 
31 31
 # Start client adapter.
32  
-$IMAGE_BASE/scripts/run-client-adapter.sh -s client-adapter -g s4 \
33  
--x -d $IMAGE_BASE/s4-core/conf/default/client-stub-conf.xml &
  32
+$S4_IMAGE/scripts/run-client-adapter.sh -s client-adapter -g s4 \
  33
+-x -d $S4_IMAGE/s4-core/conf/default/client-stub-conf.xml &
34 34
 
35 35
 # Inject events.
36  
-perl  $IMAGE_BASE/s4-driver/scripts/inject.pl RawSpeech \
37  
-io.s4.example.speech01.Speech < $IMAGE_BASE/testinput/speech.in 
  36
+perl  $S4_IMAGE/s4-driver/scripts/inject.pl RawSpeech \
  37
+io.s4.example.speech01.Speech < $S4_IMAGE/testinput/speech.in 
38 38
 
39  
-perl  $IMAGE_BASE/s4-driver/scripts/inject.pl RawSentence \
40  
-io.s4.example.speech01.Sentence < $IMAGE_BASE/testinput/sentence.in
  39
+perl  $S4_IMAGE/s4-driver/scripts/inject.pl RawSentence \
  40
+io.s4.example.speech01.Sentence < $S4_IMAGE/testinput/sentence.in
41 41
 
42 42
 
43 43
 ## Injecting Events with a Java Client
@@ -45,11 +45,11 @@ io.s4.example.speech01.Sentence < $IMAGE_BASE/testinput/sentence.in
45 45
 # Follow same steps as before to start S4 server and client adapter.
46 46
 
47 47
 # Inject events.
48  
-$IMAGE_BASE/s4-driver/scripts/inject.sh localhost 2334 RawSpeech \
49  
-io.s4.example.speech01.Speech < $IMAGE_BASE/testinput/speech.in 
  48
+$S4_IMAGE/s4-driver/scripts/inject.sh localhost 2334 RawSpeech \
  49
+io.s4.example.speech01.Speech < $S4_IMAGE/testinput/speech.in 
50 50
 
51  
-$IMAGE_BASE/s4-driver/scripts/inject.sh localhost 2334 RawSentence \
52  
-io.s4.example.speech01.Sentence < $IMAGE_BASE/testinput/sentence.in
  51
+$S4_IMAGE/s4-driver/scripts/inject.sh localhost 2334 RawSentence \
  52
+io.s4.example.speech01.Sentence < $S4_IMAGE/testinput/sentence.in
53 53
 
54 54
 
55 55
 ## Receiving Events
@@ -58,7 +58,7 @@ io.s4.example.speech01.Sentence < $IMAGE_BASE/testinput/sentence.in
58 58
 
59 59
 # Start a reader client.
60 60
 
61  
-perl $IMAGE_BASE/s4-driver/scripts/read.pl  \
  61
+perl $S4_IMAGE/s4-driver/scripts/read.pl  \
62 62
   '{ 
63 63
       readMode => "select",
64 64
       readInclude => ["SentenceJoined"]
@@ -67,28 +67,28 @@ perl $IMAGE_BASE/s4-driver/scripts/read.pl  \
67 67
 # In a different window, inject messages like in the previous section.
68 68
 
69 69
 # Remember to initialize the environment variables in the new shell.
70  
-export IMAGE_BASE=`pwd`'/build/s4-image'
71  
-export PYTHONPATH=${IMAGE_BASE}/s4-driver/lib/python
72  
-export PERLLIB=${IMAGE_BASE}/s4-driver/lib/perl
  70
+export S4_IMAGE=`pwd`'/build/s4-image'
  71
+export PYTHONPATH=${S4_IMAGE}/s4-driver/lib/python
  72
+export PERLLIB=${S4_IMAGE}/s4-driver/lib/perl
73 73
 
74 74
 # Inject events.
75  
-perl  $IMAGE_BASE/s4-driver/scripts/inject.pl RawSpeech \
76  
-io.s4.example.speech01.Speech < $IMAGE_BASE/testinput/speech.in 
  75
+perl  $S4_IMAGE/s4-driver/scripts/inject.pl RawSpeech \
  76
+io.s4.example.speech01.Speech < $S4_IMAGE/testinput/speech.in 
77 77
 
78  
-perl  $IMAGE_BASE/s4-driver/scripts/inject.pl RawSentence \
79  
-io.s4.example.speech01.Sentence < $IMAGE_BASE/testinput/sentence.in
  78
+perl  $S4_IMAGE/s4-driver/scripts/inject.pl RawSentence \
  79
+io.s4.example.speech01.Sentence < $S4_IMAGE/testinput/sentence.in
80 80
 
81 81
 
82 82
 ## Request-Response
83 83
 
84 84
 # Example 1: query the prototype of the joiner (SentenceJoinPE) 
85 85
 # in the speech02 application.
86  
-python $IMAGE_BASE/s4-driver/scripts/request.py '#sentenceJoinPE' \
87  
-'io.s4.message.PrototypeRequest' < $IMAGE_BASE/testinput/proto-query 
  86
+python $S4_IMAGE/s4-driver/scripts/request.py '#sentenceJoinPE' \
  87
+'io.s4.message.PrototypeRequest' < $S4_IMAGE/testinput/proto-query 
88 88
 
89 89
 # Example 2: request to a single PE from 
90  
- python $IMAGE_BASE/s4-driver/scripts/request.py '#sentenceJoinPE' \
91  
-'io.s4.message.SinglePERequest' < $IMAGE_BASE/testinput/pe-query 
  90
+ python $S4_IMAGE/s4-driver/scripts/request.py '#sentenceJoinPE' \
  91
+'io.s4.message.SinglePERequest' < $S4_IMAGE/testinput/pe-query 
92 92
 
93 93
 -------
94 94
 
BIN  gradle/wrapper/gradle-wrapper.jar
Binary file not shown
4  gradle/wrapper/gradle-wrapper.properties
... ...
@@ -1,6 +1,6 @@
1  
-#Mon Mar 07 11:53:07 PST 2011
  1
+#Fri May 27 15:18:16 PDT 2011
2 2
 distributionBase=GRADLE_USER_HOME
3 3
 distributionPath=wrapper/dists
4 4
 zipStoreBase=GRADLE_USER_HOME
5 5
 zipStorePath=wrapper/dists
6  
-distributionUrl=http\://gradle.artifactoryonline.com/gradle/distributions/gradle-1.0-milestone-1-bin.zip
  6
+distributionUrl=http\://repo.gradle.org/gradle/distributions/gradle-1.0-milestone-3-bin.zip
31  s4-core/README.md
Source Rendered
... ...
@@ -1,31 +0,0 @@
1  
-S4 Core Classes
2  
-===============
3  
-
4  
-Introduction
5  
-------------
6  
-This is a component of the S4 streaming system. For more information, see [s4.io](http://s4.io)
7  
-
8  
-Requirements
9  
-------------
10  
-
11  
-* Linux
12  
-* Java 1.6
13  
-* Maven
14  
-* S4 Communication Layer
15  
-
16  
-Build Instructions
17  
-------------------
18  
-
19  
-1. First build and install the comm package in your Maven repository.
20  
-
21  
-2. Kryo, Reflectasm, and minlog must be installed to your local Maven repository manually.
22  
-   The jars are present in lib/ within this project. To install, run the following commands:
23  
-
24  
-        mvn install:install-file -DgroupId=com.esotericsoftware -DartifactId=kryo -Dversion=1.01 -Dpackaging=jar -Dfile=lib/kryo-1.01.jar
25  
-        mvn install:install-file -DgroupId=com.esotericsoftware -DartifactId=reflectasm -Dversion=0.8 -Dpackaging=jar -Dfile=lib/reflectasm-0.8.jar
26  
-        mvn install:install-file -DgroupId=com.esotericsoftware -DartifactId=minlog -Dversion=1.2 -Dpackaging=jar -Dfile=lib/minlog-1.2.jar
27  
-
28  
-3. Build and install using Maven
29  
-
30  
-        mvn assembly:assembly install
31  
-
33  s4-core/src/main/java/io/s4/MainApp.java
@@ -15,8 +15,8 @@
15 15
  */
16 16
 package io.s4;
17 17
 
  18
+import io.s4.processor.AbstractPE;
18 19
 import io.s4.processor.PEContainer;
19  
-import io.s4.processor.ProcessingElement;
20 20
 import io.s4.util.S4Util;
21 21
 import io.s4.util.Watcher;
22 22
 import io.s4.util.clock.Clock;
@@ -201,9 +201,9 @@ public static void main(String args[]) throws Exception {
201 201
         coreContext = new FileSystemXmlApplicationContext(coreConfigFileUrls, coreContext);
202 202
         ApplicationContext context = coreContext;        
203 203
         
204  
-        Clock s4Clock = (Clock) context.getBean("clock");
205  
-        if (s4Clock instanceof EventClock && seedTime > 0) {
206  
-            EventClock s4EventClock = (EventClock)s4Clock;
  204
+        Clock clock = (Clock) context.getBean("clock");
  205
+        if (clock instanceof EventClock && seedTime > 0) {
  206
+            EventClock s4EventClock = (EventClock)clock;
207 207
             s4EventClock.updateTime(seedTime);
208 208
             System.out.println("Intializing event clock time with seed time " + s4EventClock.getCurrentTime());
209 209
         }
@@ -236,26 +236,19 @@ public static void main(String args[]) throws Exception {
236 236
                                                           context);
237 237
             // attach any beans that implement ProcessingElement to the PE
238 238
             // Container
239  
-            String[] processingElementBeanNames = context.getBeanNamesForType(ProcessingElement.class);
  239
+            String[] processingElementBeanNames = context.getBeanNamesForType(AbstractPE.class);
240 240
             for (String processingElementBeanName : processingElementBeanNames) {
241  
-                Object bean = context.getBean(processingElementBeanName);
242  
-                try {
243  
-                    Method getS4ClockMethod = bean.getClass().getMethod("getS4Clock");
244  
-    
245  
-                    if (getS4ClockMethod.getReturnType().equals(Clock.class)) {
246  
-                        if (getS4ClockMethod.invoke(bean) == null) {
247  
-                            Method setS4ClockMethod = bean.getClass().getMethod("setS4Clock", Clock.class);
248  
-                            setS4ClockMethod.invoke(bean, coreContext.getBean("clock"));
249  
-                        }
250  
-                    }
251  
-                }
252  
-                catch (NoSuchMethodException mnfe) {
253  
-                    // acceptable
  241
+                AbstractPE bean = (AbstractPE) context.getBean(processingElementBeanName);
  242
+                bean.setClock(clock);
  243
+                
  244
+                // if the application did not specify an id, use the Spring bean name
  245
+                if (bean.getId() == null) {
  246
+                    bean.setId(processingElementBeanName);
254 247
                 }
255 248
                 System.out.println("Adding processing element with bean name "
256 249
                         + processingElementBeanName + ", id "
257  
-                        + ((ProcessingElement) bean).getId());
258  
-                peContainer.addProcessor((ProcessingElement) bean);
  250
+                        + ((AbstractPE) bean).getId());
  251
+                peContainer.addProcessor((AbstractPE) bean);
259 252
             }
260 253
         }  
261 254
     }
18  s4-core/src/main/java/io/s4/dispatcher/partitioner/RoundRobinPartitioner.java
@@ -16,13 +16,29 @@
16 16
 package io.s4.dispatcher.partitioner;
17 17
 
18 18
 import java.util.ArrayList;
  19
+import java.util.HashSet;
19 20
 import java.util.List;
  21
+import java.util.Set;
20 22
 
21 23
 public class RoundRobinPartitioner implements Partitioner {
22 24
     private int counter = 0;
  25
+    private Set<String> streamNameSet;
  26
+
  27
+    public void setStreamNames(String[] streamNames) {
  28
+        streamNameSet = new HashSet<String>(streamNames.length);
  29
+        for (String eventType : streamNames) {
  30
+            streamNameSet.add(eventType);
  31
+        }
  32
+    }
23 33
 
24 34
     @Override
25  
-    public List<CompoundKeyInfo> partition(String streamName, Object event, int partitionCount) {
  35
+    public List<CompoundKeyInfo> partition(String streamName, Object event,
  36
+            int partitionCount) {
  37
+
  38
+        if (streamName != null && streamNameSet != null
  39
+                && !streamNameSet.contains(streamName)) {
  40
+            return null;
  41
+        }
26 42
 
27 43
         CompoundKeyInfo partitionInfo = new CompoundKeyInfo();
28 44
         int partitionId = 0;
2  s4-core/src/main/java/io/s4/ft/SafeKeeperId.java
... ...
@@ -1,7 +1,5 @@
1 1
 package io.s4.ft;
2 2
 
3  
-import io.s4.processor.ProcessingElement;
4  
-
5 3
 import java.util.regex.Matcher;
6 4
 import java.util.regex.Pattern;
7 5
 
4  s4-core/src/main/java/io/s4/message/SinglePERequest.java
@@ -18,7 +18,7 @@
18 18
 import io.s4.dispatcher.partitioner.CompoundKeyInfo;
19 19
 import io.s4.dispatcher.partitioner.Hasher;
20 20
 import io.s4.dispatcher.partitioner.KeyInfo;
21  
-import io.s4.processor.ProcessingElement;
  21
+import io.s4.processor.AbstractPE;
22 22
 import io.s4.util.MethodInvoker;
23 23
 
24 24
 import java.util.ArrayList;
@@ -83,7 +83,7 @@ public String toString() {
83 83
      * @param pe
84 84
      * @return Response object.
85 85
      */
86  
-    public Response evaluate(ProcessingElement pe) {
  86
+    public Response evaluate(AbstractPE pe) {
87 87
 
88 88
         HashMap<String, Object> results = new HashMap<String, Object>();
89 89
         HashMap<String, String> exceptions = new HashMap<String, String>();
49  s4-core/src/main/java/io/s4/processor/AbstractPE.java
@@ -57,7 +57,7 @@
57 57
  * {@link AbstractPE#setOutputFrequencyByEventCount} and
58 58
  * {@link AbstractPE#setOutputFrequencyByTimeBoundary}.
59 59
  */
60  
-public abstract class AbstractPE implements ProcessingElement {
  60
+public abstract class AbstractPE implements Cloneable {
61 61
     public static enum FrequencyType {
62 62
         TIMEBOUNDARY("timeboundary"), EVENTCOUNT("eventcount");
63 63
 
@@ -84,7 +84,7 @@ public String getName() {
84 84
         }
85 85
     }
86 86
 
87  
-    transient private Clock s4Clock;
  87
+    transient private Clock clock;
88 88
     // FIXME replaces monitor wait on AbstractPE, for triggering possible extra
89 89
     // thread when checkpointing activated
90 90
     transient private CountDownLatch s4ClockSetSignal = new CountDownLatch(1);
@@ -103,6 +103,7 @@ public String getName() {
103 103
     transient private int outputsBeforePause = -1;
104 104
     transient private long pauseTimeInMillis;
105 105
     transient private boolean logPauses = false;
  106
+    private String id;
106 107
     transient private String initMethod = null;
107 108
     transient protected SchemaContainer schemaContainer = new SchemaContainer();
108 109
     
@@ -149,27 +150,32 @@ public void setLogPauses(boolean logPauses) {
149 150
         this.logPauses = logPauses;
150 151
     }
151 152
 
152  
-    public void setS4Clock(Clock s4Clock) {
153  
-        this.s4Clock = s4Clock;
154  
-        if (this.s4Clock != null) {
155  
-            this.s4ClockSetSignal.countDown();
156  
-        }
  153
+    public String getId() {
  154
+        return id;
157 155
     }
158 156
 
159  
-    /**
160  
-     * The name of a method to be used as an initializer. The method will be
161  
-     * called after the object is cloned from the prototype PE.
162  
-     */
163  
-    public void setInitMethod(String initMethod) {
164  
-        this.initMethod = initMethod;
  157
+    public void setId(String id) {
  158
+        this.id = id;
165 159
     }
166 160
 
167  
-    public String getInitMethod() {
168  
-        return this.initMethod;
  161
+    public void setClock(Clock clock) {
  162
+        synchronized (this) {
  163
+            this.clock = clock;
  164
+            this.notify();
  165
+        }
169 166
     }
170 167
 
171  
-    public Clock getS4Clock() {
172  
-        return s4Clock;
  168
+    /**
  169
+     * This method will be called after the object is cloned from the
  170
+     * prototype PE. The concrete PE class should override this if
  171
+     * it has any special set-up requirements.
  172
+     */
  173
+    public void initInstance() {
  174
+       // default implementation does nothing.
  175
+    }
  176
+        
  177
+    public Clock getClock() {
  178
+        return clock;
173 179
     }
174 180
 
175 181
     public AbstractPE() {
@@ -245,7 +251,7 @@ public void execute(String streamName, CompoundKeyInfo compoundKeyInfo,
245 251
     }
246 252
 
247 253
     public long getCurrentTime() {
248  
-        return s4Clock.getCurrentTime();
  254
+        return clock.getCurrentTime();
249 255
     }
250 256
 
251 257
     /**
@@ -678,7 +684,7 @@ public long getFrequencyOffset() {
678 684
         }
679 685
 
680 686
         public void run() {
681  
-            if (s4Clock == null) {
  687
+            if (clock == null) {
682 688
                 try {
683 689
                     s4ClockSetSignal.await();
684 690
                 } catch (InterruptedException e) {
@@ -701,9 +707,8 @@ public void run() {
701 707
                 long currentBoundary = (currentTime / frequencyInMillis)
702 708
                         * frequencyInMillis;
703 709
                 long nextBoundary = currentBoundary + frequencyInMillis;
704  
-                currentTime = s4Clock.waitForTime(nextBoundary
705  
-                        + (getFrequencyOffset() * 1000));
706  
-
  710
+                currentTime = clock.waitForTime(nextBoundary
  711
+                        + (outputFrequencyOffset * 1000));
707 712
                 if (lookupTable != null) {
708 713
                     Set peKeys = lookupTable.keySet();
709 714
                     for (Iterator it = peKeys.iterator(); it.hasNext();) {
2  s4-core/src/main/java/io/s4/processor/ControlEventProcessor.java
@@ -60,7 +60,7 @@ protected void execute(EventWrapper e, PrototypeWrapper p) {
60 60
 
61 61
             String keyVal = keyInfo.getCompoundValue();
62 62
 
63  
-            ProcessingElement pe = p.lookupPE(keyVal);
  63
+            AbstractPE pe = p.lookupPE(keyVal);
64 64
 
65 65
             Response response = ((SinglePERequest) event).evaluate(pe);
66 66
             String stream = response.getRInfo().getStream();
9  s4-core/src/main/java/io/s4/processor/JoinPE.java
@@ -35,19 +35,10 @@
35 35
     private Map<String, Object> eventsToJoin;
36 36
     private EventDispatcher dispatcher;
37 37
     private Monitor monitor;
38  
-    private String id = "JoinPE";
39 38
     private String outputStreamName;
40 39
     private String outputClassName;
41 40
     private Class<?> outputClass;
42 41
 
43  
-    public String getId() {
44  
-        return id;
45  
-    }
46  
-
47  
-    public void setId(String id) {
48  
-        this.id = id;
49  
-    }
50  
-
51 42
     public void setDispatcher(EventDispatcher dispatcher) {
52 43
         this.dispatcher = dispatcher;
53 44
     }
22  s4-core/src/main/java/io/s4/processor/PEContainer.java
@@ -49,7 +49,7 @@
49 49
     BlockingQueue<EventWrapper> workQueue;
50 50
     private List<PrototypeWrapper> prototypeWrappers = new ArrayList<PrototypeWrapper>();
51 51
     private Monitor monitor;
52  
-    private Clock s4Clock;
  52
+    private Clock clock;
53 53
     private int maxQueueSize = 1000;
54 54
     private boolean trackByKey;
55 55
     private Map<String, Integer> countByEventType = Collections.synchronizedMap(new HashMap<String, Integer>());
@@ -65,12 +65,12 @@ public void setMonitor(Monitor monitor) {
65 65
         this.monitor = monitor;
66 66
     }
67 67
 
68  
-    public void setS4Clock(Clock s4Clock) {
69  
-        this.s4Clock = s4Clock;
  68
+    public void setClock(Clock s4Clock) {
  69
+        this.clock = s4Clock;
70 70
     }
71 71
 
72  
-    public Clock getS4Clock() {
73  
-        return s4Clock;
  72
+    public Clock getClock() {
  73
+        return clock;
74 74
     }
75 75
 
76 76
     public void setTrackByKey(boolean trackByKey) {
@@ -81,15 +81,15 @@ public void setSafeKeeper(SafeKeeper sk) {
81 81
 		this.safeKeeper = sk;
82 82
 	}
83 83
 
84  
-    public void addProcessor(ProcessingElement processor) {
  84
+    public void addProcessor(AbstractPE processor) {
85 85
         System.out.println("adding pe: " + processor);
86  
-        PrototypeWrapper pw = new PrototypeWrapper(processor, s4Clock);
  86
+        PrototypeWrapper pw = new PrototypeWrapper(processor, clock);
87 87
         pw.setSafeKeeper(safeKeeper);
88 88
         prototypeWrappers.add(pw);
89 89
         adviceLists.add(pw.advise());
90 90
     }
91 91
 
92  
-    public void setProcessors(ProcessingElement[] processors) {
  92
+    public void setProcessors(AbstractPE[] processors) {
93 93
         // prototypeWrappers = new ArrayList<PrototypeWrapper>();
94 94
 
95 95
         for (int i = 0; i < processors.length; i++) {
@@ -188,8 +188,8 @@ public void run() {
188 188
             EventWrapper eventWrapper = null;
189 189
             try {
190 190
                 eventWrapper = workQueue.take();
191  
-                if (s4Clock instanceof EventClock) {
192  
-                    EventClock eventClock = (EventClock) s4Clock;
  191
+                if (clock instanceof EventClock) {
  192
+                    EventClock eventClock = (EventClock) clock;
193 193
                     eventClock.update(eventWrapper);
194 194
                     // To what time to update the clock
195 195
                 }
@@ -335,7 +335,7 @@ private void handleCheckpointingOrRecovery(EventWrapper eventWrapper) {
335 335
 
336 336
     }
337 337
 
338  
-    private void invokePE(ProcessingElement pe, EventWrapper eventWrapper,
  338
+    private void invokePE(AbstractPE pe, EventWrapper eventWrapper,
339 339
                           CompoundKeyInfo compoundKeyInfo) {
340 340
         try {
341 341
             long startTime = System.currentTimeMillis();
10  s4-core/src/main/java/io/s4/processor/PrintEventPE.java
@@ -19,16 +19,6 @@
19 19
 
20 20
 public class PrintEventPE extends AbstractPE {
21 21
 
22  
-    private String id = "PrintEventPE";
23  
-
24  
-    public String getId() {
25  
-        return id;
26  
-    }
27  
-
28  
-    public void setId(String id) {
29  
-        this.id = id;
30  
-    }
31  
-
32 22
     @Override
33 23
     public void output() {
34 24
         // TODO Auto-generated method stub
24  s4-core/src/main/java/io/s4/processor/PrototypeWrapper.java
@@ -28,7 +28,7 @@
28 28
 public class PrototypeWrapper {
29 29
 
30 30
     private static Logger logger = Logger.getLogger(PrototypeWrapper.class);
31  
-    private ProcessingElement prototype;
  31
+    private AbstractPE prototype;
32 32
     Persister lookupTable;
33 33
 	SafeKeeper safeKeeper;
34 34
 
@@ -36,7 +36,7 @@ public String getId() {
36 36
         return prototype.getId();
37 37
     }
38 38
 
39  
-    public PrototypeWrapper(ProcessingElement prototype, Clock s4Clock) {
  39
+    public PrototypeWrapper(AbstractPE prototype, Clock s4Clock) {
40 40
         this.prototype = prototype;
41 41
         lookupTable = new ConMapPersister(s4Clock);
42 42
         // TODO lookup table with PEIds
@@ -70,21 +70,17 @@ public PrototypeWrapper(ProcessingElement prototype, Clock s4Clock) {
70 70
      *            key value
71 71
      * @return PE corresponding to keyValue.
72 72
      */
73  
-    public ProcessingElement getPE(String keyValue) {
74  
-        ProcessingElement pe = null;
  73
+    public AbstractPE getPE(String keyValue) {
  74
+        AbstractPE pe = null;
75 75
         try {
76  
-            pe = (ProcessingElement) lookupTable.get(keyValue);
  76
+            pe = (AbstractPE) lookupTable.get(keyValue);
77 77
             if (pe == null) {
78  
-                pe = (ProcessingElement) prototype.clone();
  78
+                pe = (AbstractPE) prototype.clone();
79 79
                 if (pe instanceof AbstractPE) {
80 80
                     // Logger.getLogger("s4").info("injecting safekeeper");
81 81
 					((AbstractPE) pe).setSafeKeeper(safeKeeper);
82 82
 				}
83  
-                //invoke the initialization method if it has been specified
84  
-                if (pe.getInitMethod() != null) {
85  
-                   Method initMethod = pe.getClass().getMethod(pe.getInitMethod(), new Class[0]);
86  
-                   initMethod.invoke(pe, (new Object[0]));
87  
-                }
  83
+                pe.initInstance();
88 84
 
89 85
             }
90 86
             // update the last update time on the entry
@@ -106,11 +102,11 @@ public ProcessingElement getPE(String keyValue) {
106 102
      * @return PE corresponding to keyValue, if such a PE exists. Null
107 103
      *         otherwise.
108 104
      */
109  
-    public ProcessingElement lookupPE(String keyValue) {
110  
-        ProcessingElement pe = null;
  105
+    public AbstractPE lookupPE(String keyValue) {
  106
+        AbstractPE pe = null;
111 107
 
112 108
         try {
113  
-            pe = (ProcessingElement) lookupTable.get(keyValue);
  109
+            pe = (AbstractPE) lookupTable.get(keyValue);
114 110
 
115 111
         } catch (Exception e) {
116 112
             logger.error("exception when looking up pe for key:" + keyValue, e);
9  s4-core/src/main/java/io/s4/processor/ReroutePE.java
@@ -30,17 +30,8 @@
30 30
     private EventDispatcher dispatcher;
31 31
     private Transformer[] transformers = new Transformer[0];
32 32
     // private List<EventAdvice> keys;
33  
-    private String id = "ReroutePE";
34 33
     private String outputStreamName;
35 34
 
36  
-    public String getId() {
37  
-        return id;
38  
-    }
39  
-
40  
-    public void setId(String id) {
41  
-        this.id = id;
42  
-    }
43  
-
44 35
     public void setDispatcher(EventDispatcher dispatcher) {
45 36
         this.dispatcher = dispatcher;
46 37
     }
9  s4-core/src/main/java/io/s4/processor/SimpleCountingPE.java
@@ -26,15 +26,6 @@
26 26
     private int persistTime;
27 27
     private String keyPrefix = "s4:counter";
28 28
     private boolean dirty = false;
29  
-    private String id = "SimpleCountingPE";
30  
-
31  
-    public String getId() {
32  
-        return id;
33  
-    }
34  
-
35  
-    public void setId(String id) {
36  
-        this.id = id;
37  
-    }
38 29
 
39 30
     public void setClearOnOutput(boolean clearOnOutput) {
40 31
         this.clearOnOutput = clearOnOutput;
0  s4-core/src/main/resources/s4-apps/README → s4-core/src/main/resources/s4-apps/.gitignore
File renamed without changes
2  s4-core/src/main/resources/s4-core/conf/default/client-adapter-conf.xml
@@ -2,7 +2,7 @@
2 2
 <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.springframework.org/schema/beans              http://www.springframework.org/schema/beans/spring-beans-2.0.xsd">
3 3
   <bean id="propertyConfigurer" class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
4 4
     <property name="location">
5  
-      <value>classpath:adapter.properties</value>
  5
+      <value>classpath:client-adapter.properties</value>
6 6
     </property>
7 7
     <property name="ignoreUnresolvablePlaceholders" value="true"/>
8 8
   </bean> 
2  s4-core/src/main/resources/s4-core/conf/default/client-stub-conf.xml
@@ -2,7 +2,7 @@
2 2
 <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.springframework.org/schema/beans              http://www.springframework.org/schema/beans/spring-beans-2.0.xsd">
3 3
   <bean id="propertyConfigurer" class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
4 4
     <property name="location">
5  
-      <value>classpath:adapter.properties</value>
  5
+      <value>classpath:client-adapter.properties</value>
6 6
     </property>
7 7
     <property name="ignoreUnresolvablePlaceholders" value="true"/>
8 8
   </bean> 
6  s4-core/src/main/resources/s4-core/conf/default/clusters.xml
@@ -7,12 +7,6 @@
7 7
       <taskId>s4node-0</taskId>
8 8
     </node>  
9 9
   </cluster>
10  
-  <cluster name="s4" type="adapter" mode="unicast">
11  
-    <node>
12  
-      <machine>localhost</machine>
13  
-      <taskId>adapter-0</taskId>
14  
-    </node>
15  
-  </cluster>
16 10
   <cluster name="client-adapter" type="s4" mode="unicast">
17 11
     <node>
18 12
       <partition>0</partition>
26  s4-core/src/main/resources/s4-core/conf/default/log4j.xml
... ...
@@ -1,6 +1,18 @@
1 1
 <?xml version="1.0" encoding="UTF-8" ?>
2 2
 <!DOCTYPE log4j:configuration SYSTEM "log4j.dtd">
3 3
 <log4j:configuration xmlns:log4j="http://jakarta.apache.org/log4j/">
  4
+  <appender name="R" class="org.apache.log4j.DailyRollingFileAppender">
  5
+    <param name="File" value="${log_loc}/s4-core/s4-core_${instanceId}.log" />
  6
+    <layout class="org.apache.log4j.PatternLayout"> 
  7
+      <param name="ConversionPattern" value="%d %c %p (%F:%L) %m%n"/>
  8
+    </layout> 
  9
+  </appender> 
  10
+  <appender name="S" class="org.apache.log4j.DailyRollingFileAppender">
  11
+    <param name="File" value="${log_loc}/s4-core/s4-core_${instanceId}.mon" />
  12
+    <layout class="org.apache.log4j.PatternLayout">
  13
+      <param name="ConversionPattern" value="%d %c %p (%F:%L) %m%n"/>
  14
+    </layout>
  15
+  </appender>
4 16
   <logger name="com.yahoo" additivity="false">
5 17
         <level value="info"/>
6 18
         <appender-ref ref="R"/>
@@ -25,22 +37,10 @@
25 37
         <level value="info"/>
26 38
         <appender-ref ref="S"/>
27 39
   </logger>
28  
-  <logger name="s4-ft">
  40
+  <logger name="s4-ft" additivity="false">
29 41
         <level value="info"/>
30 42
         <appender-ref ref="R"/>
31 43
   </logger>
32  
-  <appender name="R" class="org.apache.log4j.DailyRollingFileAppender">
33  
-    <param name="File" value="${log_loc}/s4-core/s4-core_${instanceId}.log" />
34  
-    <layout class="org.apache.log4j.PatternLayout"> 
35  
-      <param name="ConversionPattern" value="%d %c %p (%F:%L) %m%n"/>
36  
-    </layout> 
37  
-  </appender> 
38  
-  <appender name="S" class="org.apache.log4j.DailyRollingFileAppender">
39  
-    <param name="File" value="${log_loc}/s4-core/s4-core_${instanceId}.mon" />
40  
-    <layout class="org.apache.log4j.PatternLayout">
41  
-      <param name="ConversionPattern" value="%d %c %p (%F:%L) %m%n"/>
42  
-    </layout>
43  
-  </appender>
44 44
   <root> 
45 45
     <priority value ="info" /> 
46 46
     <appender-ref ref="R" /> 
2  s4-core/src/main/resources/s4-core/conf/default/s4-core-conf.xml
@@ -101,7 +101,7 @@
101 101
     <property name="maxQueueSize" value="${pe_container_max_queue_size}"/>
102 102
     <property name="monitor" ref="monitor"/>
103 103
     <property name="trackByKey" value="true"/>
104  
-    <property name="s4Clock" ref="clock"/>
  104
+    <property name="clock" ref="clock"/>
105 105
     <property name="controlEventProcessor" ref="ctrlHandler"/>
106 106
   </bean>
107 107
 
41  s4-core/src/main/resources/s4-core/conf/dynamic/adapter-conf.xml
... ...
@@ -0,0 +1,41 @@
  1
+<?xml version="1.0" encoding="UTF-8"?>
  2
+<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.springframework.org/schema/beans              http://www.springframework.org/schema/beans/spring-beans-2.0.xsd">
  3
+  <bean id="propertyConfigurer" class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
  4
+    <property name="location">
  5
+      <value>classpath:adapter.properties</value>
  6
+    </property>
  7
+    <property name="ignoreUnresolvablePlaceholders" value="true"/>
  8
+  </bean> 
  9
+  
  10
+    <bean id="commLayerEmitter" class="io.s4.emitter.CommLayerEmitter" init-method="init">
  11
+    <property name="clusterManagerAddress" value="${zk_address}"/>
  12
+    <property name="serDeser" ref="serDeser"/>
  13
+    <property name="appName" value="${appName}"/>
  14
+    <property name="listenerAppName" value="${listenerAppName}"/>
  15
+    <property name="monitor" ref="monitor"/>
  16
+  </bean>
  17
+
  18
+  <bean id="serDeser" class="io.s4.serialize.KryoSerDeser"/>
  19
+
  20
+  <bean id="monitor" class="io.s4.logger.Log4jMonitor" lazy-init="true" init-method="init">
  21
+    <property name="flushInterval" value="30"/>
  22
+    <property name="loggerName" value="monitor"/>
  23
+  </bean>
  24
+
  25
+  <bean id="dummyPartitioner" class="io.s4.dispatcher.partitioner.DummyPartitioner"/>
  26
+
  27
+  <bean id="dispatcher" class="io.s4.dispatcher.Dispatcher" init-method="init">
  28
+    <property name="partitioners">
  29
+      <list>
  30
+        <ref bean="dummyPartitioner"/>
  31
+      </list>
  32
+    </property>
  33
+    <property name="eventEmitter" ref="commLayerEmitter"/>
  34
+    <property name="loggerName" value="s4"/>
  35
+  </bean>
  36
+
  37
+  <bean id="adapter" class="io.s4.adapter.Adapter"
  38
+        init-method="init">
  39
+    <property name="dispatcher" ref="dispatcher"/>
  40
+  </bean>
  41
+</beans>
64  s4-core/src/main/resources/s4-core/conf/dynamic/client-adapter-conf.xml
... ...
@@ -0,0 +1,64 @@
  1
+<?xml version="1.0" encoding="UTF-8"?>
  2
+<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.springframework.org/schema/beans              http://www.springframework.org/schema/beans/spring-beans-2.0.xsd">
  3
+  <bean id="propertyConfigurer" class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
  4
+    <property name="location">
  5
+      <value>classpath:client-adapter.properties</value>
  6
+    </property>
  7
+    <property name="ignoreUnresolvablePlaceholders" value="true"/>
  8
+  </bean> 
  9
+  
  10
+  <bean id="commLayerEmitter" class="io.s4.emitter.CommLayerEmitter" init-method="init">
  11
+    <property name="serDeser" ref="serDeser"/>
  12
+    <property name="appName" value="${appName}"/>
  13
+    <property name="listenerAppName" value="${listenerAppName}"/>
  14
+    <property name="listener" ref="rawListener"/>
  15
+    <property name="monitor" ref="monitor"/>
  16
+  </bean>
  17
+
  18
+  <bean id="serDeser" class="io.s4.serialize.KryoSerDeser"/>
  19
+
  20
+  <bean id="rawListener" class="io.s4.listener.CommLayerListener" init-method="init">
  21
+    <property name="serDeser" ref="serDeser"/>
  22
+    <property name="clusterManagerAddress" value="${zk_address}"/>
  23
+    <!-- <property name="clusterType" value="ADAPTER"/> -->
  24
+    <property name="appName" value="${appName}"/>
  25
+    <property name="maxQueueSize" value="${listener_max_queue_size}"/>
  26
+    <property name="monitor" ref="monitor"/>
  27
+  </bean>
  28
+
  29
+  <bean id="eventListener" class="io.s4.collector.EventListener" init-method="init">
  30
+    <property name="rawListener" ref="rawListener"/>
  31
+    <property name="monitor" ref="monitor"/>
  32
+  </bean>
  33
+
  34
+  <bean id="monitor" class="io.s4.logger.Log4jMonitor" lazy-init="true" init-method="init">
  35
+    <property name="flushInterval" value="30"/>
  36
+    <property name="loggerName" value="monitor"/>
  37
+  </bean>
  38
+
  39
+  <bean id="dummyPartitioner" class="io.s4.dispatcher.partitioner.DummyPartitioner"/>
  40
+
  41
+  <bean id="hasher" class="io.s4.dispatcher.partitioner.DefaultHasher"/>
  42
+  
  43
+  <bean id="genericPartitioner" class="io.s4.dispatcher.partitioner.DefaultPartitioner">
  44
+    <property name="hasher" ref="hasher"/>
  45
+    <property name="debug" value="false"/>
  46
+  </bean>
  47
+
  48
+
  49
+  <bean id="dispatcher" class="io.s4.dispatcher.Dispatcher" init-method="init">
  50
+    <property name="partitioners">
  51
+      <list>
  52
+        <ref bean="genericPartitioner"/>
  53
+      </list>
  54
+    </property>
  55
+    <property name="eventEmitter" ref="commLayerEmitter"/>
  56
+    <property name="loggerName" value="s4"/>
  57
+  </bean>
  58
+
  59
+  <bean id="client_adapter" class="io.s4.client.Adapter"
  60
+        init-method="init">
  61
+    <property name="dispatcher" ref="dispatcher"/>
  62
+    <property name="clusterEventListener" ref="eventListener"/>
  63
+  </bean>
  64
+</beans>
13  s4-core/src/main/resources/s4-core/conf/dynamic/client-stub-conf.xml
... ...
@@ -0,0 +1,13 @@
  1
+<?xml version="1.0" encoding="UTF-8"?>
  2
+<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.springframework.org/schema/beans              http://www.springframework.org/schema/beans/spring-beans-2.0.xsd">
  3
+  <bean id="propertyConfigurer" class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
  4
+    <property name="location">
  5
+      <value>classpath:client-adapter.properties</value>
  6
+    </property>
  7
+    <property name="ignoreUnresolvablePlaceholders" value="true"/>
  8
+  </bean> 
  9
+  
  10
+  <bean id="genericStub" class="io.s4.client.GenericJsonClientStub" init-method="init">
  11
+    <property name="connectionPort" value="2334"/>
  12
+  </bean>
  13
+</beans>
18  s4-core/src/main/resources/s4-core/conf/dynamic/clusters.xml
... ...
@@ -0,0 +1,18 @@
  1
+<config version="-1">
  2
+  <cluster name="s4" type="s4" mode="unicast">
  3
+    <node>
  4
+      <partition>0</partition>
  5
+      <machine>localhost</machine>
  6
+      <port>5077</port>
  7
+      <taskId>s4node-0</taskId>
  8
+    </node>  
  9
+  </cluster>
  10
+  <cluster name="client-adapter" type="s4" mode="unicast">
  11
+    <node>
  12
+      <partition>0</partition>
  13
+      <machine>localhost</machine>
  14
+      <taskId>client-adapter-0</taskId>
  15
+      <port>6077</port>
  16
+    </node>
  17
+  </cluster>
  18
+</config>
6  s4-core/src/main/resources/s4-core/conf/dynamic/event-clock.xml
... ...
@@ -0,0 +1,6 @@
  1
+<?xml version="1.0" encoding="UTF-8"?>
  2
+<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.springframework.org/schema/beans              http://www.springframework.org/schema/beans/spring-beans-2.0.xsd">
  3
+ 
  4
+  <bean id="clock" class="io.s4.util.clock.EventClock"/>
  5
+
  6
+</beans>
48  s4-core/src/main/resources/s4-core/conf/dynamic/log4j.xml
... ...
@@ -0,0 +1,48 @@
  1
+<?xml version="1.0" encoding="UTF-8" ?>
  2
+<!DOCTYPE log4j:configuration SYSTEM "log4j.dtd">
  3
+<log4j:configuration xmlns:log4j="http://jakarta.apache.org/log4j/">
  4
+  <logger name="com.yahoo" additivity="false">
  5
+        <level value="info"/>
  6
+        <appender-ref ref="R"/>
  7
+  </logger>
  8
+  <logger name="s4" additivity="false">
  9
+        <level value="info"/>
  10
+        <appender-ref ref="R"/>
  11
+  </logger>
  12
+  <logger name="zk" additivity="false">
  13
+        <level value="info"/>
  14
+        <appender-ref ref="R"/>
  15
+  </logger>
  16
+  <logger name="dispatcher" additivity="false">
  17
+        <level value="info"/>
  18
+        <appender-ref ref="R"/>
  19
+  </logger>
  20
+  <logger name="adapter" additivity="false">
  21
+        <level value="info"/>
  22
+        <appender-ref ref="R"/>
  23
+  </logger>
  24
+  <logger name="monitor">
  25
+        <level value="info"/>
  26
+        <appender-ref ref="S"/>
  27
+  </logger>
  28
+  <appender name="R" class="org.apache.log4j.DailyRollingFileAppender">
  29
+    <param name="File" value="${log_loc}/s4-core/s4-core_${instanceId}.log" />
  30
+    <layout class="org.apache.log4j.PatternLayout"> 
  31
+      <param name="ConversionPattern" value="%d %c %p (%F:%L) %m%n"/>
  32
+    </layout> 
  33
+  </appender> 
  34
+  <appender name="S" class="org.apache.log4j.DailyRollingFileAppender">
  35
+    <param name="File" value="${log_loc}/s4-core/s4-core_${instanceId}.mon" />
  36
+    <layout class="org.apache.log4j.PatternLayout">
  37
+      <param name="ConversionPattern" value="%d %c %p (%F:%L) %m%n"/>
  38
+    </layout>
  39
+  </appender>
  40
+  <root> 
  41
+    <priority value ="info" /> 
  42
+    <appender-ref ref="R" /> 
  43
+  </root>
  44
+</log4j:configuration>
  45
+
  46
+
  47
+
  48
+
152  s4-core/src/main/resources/s4-core/conf/dynamic/s4-core-conf.xml
... ...
@@ -0,0 +1,152 @@
  1
+<?xml version="1.0" encoding="UTF-8"?>
  2
+<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.springframework.org/schema/beans              http://www.springframework.org/schema/beans/spring-beans-2.0.xsd">
  3
+  <bean id="propertyConfigurer" class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
  4
+    <property name="location">
  5
+      <value>classpath:s4-core.properties</value>
  6
+    </property>
  7
+    <property name="properties">
  8
+      <props>
  9
+        <prop key="kryoSerDeser.initialBufferSize">2048</prop>
  10
+        <prop key="kryoSerDeser.maxBufferSize">262144</prop>
  11
+      </props>
  12
+    </property>
  13
+    <property name="ignoreUnresolvablePlaceholders" value="true"/>
  14
+  </bean> 
  15
+  
  16
+  <bean id="hasher" class="io.s4.dispatcher.partitioner.DefaultHasher"/>
  17
+  
  18
+  <bean id="commLayerEmitterToAdapter" class="io.s4.emitter.CommLayerEmitter" init-method="init">
  19
+    <property name="serDeser" ref="serDeser"/>
  20
+    <property name="listener" ref="rawListener"/>
  21
+    <property name="listenerAppName" value="${adapter_app_name}"/>
  22
+    <property name="monitor" ref="monitor"/>
  23
+  </bean>
  24
+
  25
+  <bean id="commLayerEmitter" class="io.s4.emitter.CommLayerEmitter" init-method="init">