Merge pull request #1604 from hushengyue/issue-1600

Fixed issue-1600 add thread-safe protection to inputManager
siddhi-io · Jan 6, 2020 · f1ef77f · f1ef77f
2 parents 5202923 + e7e87eb
commit f1ef77f
Showing 1 changed file with 12 additions and 5 deletions.
diff --git a/modules/siddhi-core/src/main/java/io/siddhi/core/stream/input/InputManager.java b/modules/siddhi-core/src/main/java/io/siddhi/core/stream/input/InputManager.java
@@ -52,11 +52,18 @@ public InputManager(SiddhiAppContext siddhiAppContext,
  public InputHandler getInputHandler(String streamId) {
  InputHandler inputHandler = inputHandlerMap.get(streamId);
  if (inputHandler == null) {
- InputHandler newInputHandler = constructInputHandler(streamId);
- if (this.isConnected) {
- newInputHandler.connect();
+ synchronized (this) {
+ inputHandler = inputHandlerMap.get(streamId);
+ if (inputHandler == null) {
+ InputHandler newInputHandler = constructInputHandler(streamId);
+ if (this.isConnected) {
+ newInputHandler.connect();
+ }
+ return newInputHandler;
+ } else {
+ return inputHandler;
+ }
  }
- return newInputHandler;
  } else {
  return inputHandler;
  }
@@ -78,7 +85,7 @@ public synchronized void disconnect() {
  this.isConnected = false;
  }
 
- public InputHandler constructInputHandler(String streamId) {
+ public synchronized InputHandler constructInputHandler(String streamId) {
  InputHandler inputHandler = new InputHandler(streamId, inputHandlerMap.size(),
  inputEntryValve, siddhiAppContext);
  StreamJunction streamJunction = streamJunctionMap.get(streamId);