Bounding spouts

wso2 · Sep 21, 2015 · f097569 · f097569
1 parent e97eac7
commit f097569
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 8 deletions.
diff --git a/.../main/java/org/wso2/carbon/event/processor/common/storm/component/EventReceiverSpout.java b/.../main/java/org/wso2/carbon/event/processor/common/storm/component/EventReceiverSpout.java
@@ -41,7 +41,7 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.LinkedBlockingQueue;
 
 /**
  * Receive events from CEP receivers through thrift receiver and pass through
@@ -72,8 +72,7 @@ public class EventReceiverSpout extends BaseRichSpout implements StreamCallback
      * this is filled by the receiver thread of data bridge and consumed by the nextTuple which
      * runs on the worker thread of spout.
      */
-    // TODO : Make this queue a fixed size to prevent out of memory issues
-    private transient ConcurrentLinkedQueue<Event> storedEvents = null;
+    private transient LinkedBlockingQueue<Event> storedEvents = null;
 
     private SpoutOutputCollector spoutOutputCollector = null;
 
@@ -125,7 +124,7 @@ public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
     @Override
     public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
         this.spoutOutputCollector = spoutOutputCollector;
-        this.storedEvents = new ConcurrentLinkedQueue<Event>();
+        this.storedEvents = new LinkedBlockingQueue<Event>(stormDeploymentConfig.getStormSpoutBufferSize());
 
         inputThroughputProbe = new ThroughputProbe(logPrefix + "-IN", 10);
         outputThroughputProbe = new ThroughputProbe(logPrefix + " -OUT", 10);
@@ -137,7 +136,6 @@ public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector
             thisHostIp = Utils.findAddress("localhost");
             listeningPort = findPort(thisHostIp);
             TCPEventServerConfig configs = new TCPEventServerConfig(thisHostIp, listeningPort);
-            configs.setNumberOfThreads(stormDeploymentConfig.getTransportReceiverThreads());
             tcpEventServer = new TCPEventServer(configs, this, null);
             for (StreamDefinition siddhiStreamDefinition : incomingStreamDefinitions) {
                 tcpEventServer.addStreamDefinition(siddhiStreamDefinition);
@@ -187,8 +185,12 @@ public void receive(String streamId, long timestamp, Object[] eventData) {
         if (log.isDebugEnabled()) {
             log.debug(logPrefix + "Received Event: " + streamId + ":" + Arrays.deepToString(eventData) + "@" + timestamp);
         }
-        storedEvents.add(new Event(timestamp, eventData, streamId));
-        inputThroughputProbe.update();
+        try {
+            storedEvents.put(new Event(timestamp, eventData, streamId));
+            inputThroughputProbe.update();
+        } catch (InterruptedException e) {
+            //ignore
+        }
     }
 
 

diff --git a/...a/org/wso2/carbon/event/processor/core/internal/storm/SiddhiStormOutputEventListener.java b/...a/org/wso2/carbon/event/processor/core/internal/storm/SiddhiStormOutputEventListener.java
@@ -81,7 +81,6 @@ private void init() {
             thisHostIp = HostAddressFinder.findAddress("localhost");
             listeningPort = findPort(thisHostIp);
             TCPEventServerConfig configs = new TCPEventServerConfig(thisHostIp, listeningPort);
-            configs.setNumberOfThreads(stormDeploymentConfig.getTransportReceiverThreads());
             tcpEventServer = new TCPEventServer(configs, this, connectionCallback);
             tcpEventServer.start();
             executorService.execute(new Registrar());