/*
 * Copyright (c) 2016, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * WSO2 Inc. licenses this file to you under the Apache License,
 * Version 2.0 (the "License"); you may not use this file except
 * in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied. See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package io.siddhi.core.stream.input;

import io.siddhi.core.config.SiddhiAppContext;
import io.siddhi.core.event.Event;
import io.siddhi.core.exception.DefinitionNotExistException;
import io.siddhi.core.stream.StreamJunction;
import io.siddhi.query.api.definition.AbstractDefinition;

import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentMap;

/**
 * Manager class to handle {@link Event} insertion to Siddhi.
 */
public class InputManager {

    private final InputEntryValve inputEntryValve;
    private final SiddhiAppContext siddhiAppContext;
    private Map<String, InputHandler> inputHandlerMap = new LinkedHashMap<String, InputHandler>();
    private Map<String, StreamJunction> streamJunctionMap;
    private InputDistributor inputDistributor;

    public InputManager(SiddhiAppContext siddhiAppContext,
                        ConcurrentMap<String, AbstractDefinition> streamDefinitionMap,
                        ConcurrentMap<String, StreamJunction> streamJunctionMap) {
        this.streamJunctionMap = streamJunctionMap;
        this.inputDistributor = new InputDistributor();
        this.inputEntryValve = new InputEntryValve(siddhiAppContext, inputDistributor);
        this.siddhiAppContext = siddhiAppContext;
    }

    public InputHandler getInputHandler(String streamId) {
        InputHandler inputHandler = inputHandlerMap.get(streamId);
        if (inputHandler == null) {
            return constructInputHandler(streamId);
        } else {
            return inputHandler;
        }
    }

    public synchronized void connect() {
        for (InputHandler inputHandler : inputHandlerMap.values()) {
            inputHandler.connect();
        }
    }

    public synchronized void disconnect() {
        for (InputHandler inputHandler : inputHandlerMap.values()) {
            inputHandler.disconnect();
        }
        inputHandlerMap.clear();
    }

    public InputHandler constructInputHandler(String streamId) {
        InputHandler inputHandler = new InputHandler(streamId, inputHandlerMap.size(),
                inputEntryValve, siddhiAppContext);
        StreamJunction streamJunction = streamJunctionMap.get(streamId);
        if (streamJunction == null) {
            throw new DefinitionNotExistException("Stream with stream ID " + streamId + " has not been defined");
        }
        inputDistributor.addInputProcessor(streamJunctionMap.get(streamId).constructPublisher());
        inputHandlerMap.put(streamId, inputHandler);
        return inputHandler;
    }
}

@@ -1458,16 +1458,57 @@
 ributor;
+%0A    private boolean isConnected = false;
 %0A%0A    pu
@@ -2157,22 +2157,46 @@
         
-return
+InputHandler newInputHandler =
  constru
@@ -2213,32 +2213,161 @@
 dler(streamId);%0A
+            if (this.isConnected) %7B%0A                newInputHandler.connect();%0A            %7D%0A            return newInputHandler;%0A
         %7D else %7B
@@ -2565,32 +2565,65 @@
 ct();%0A        %7D%0A
+        this.isConnected = true;%0A
     %7D%0A%0A    publi
@@ -2767,32 +2767,32 @@
 ct();%0A        %7D%0A
-
         inputHan
@@ -2804,24 +2804,58 @@
 ap.clear();%0A
+        this.isConnected = false;%0A
     %7D%0A%0A    p

connect inputHandler if attached after siddhi pp is started