[org.wso2.extension.siddhi.io.http.source.HttpConnectorPortBindingListener] : siddhi: started HTTP server connector localhost:8081
[org.wso2.extension.siddhi.io.http.source.HttpSourceListener] : Source Listener has created for url http://localhost:8081/Api/public/test
[org.wso2.extension.siddhi.io.http.source.HttpSourceListener] : Source Listener has created for url http://localhost:8081/Api/public/test
ERROR [org.wso2.siddhi.core.stream.input.source.Source] - Error on 'TestExecutSionPlan'. Listener URL http://localhost:8081/Api/public/test already connectedError while connecting at Source 'http' at 'FooStream'.
org.wso2.siddhi.core.exception.SiddhiAppCreationException: Error on 'TestExecutSionPlan'. Listener URL http://localhost:8081/Api/public/test already connected
	at org.wso2.extension.siddhi.io.http.source.HttpConnectorRegistry.registerSourceListener(HttpConnectorRegistry.java:143)

@App:name('testSiddhi')

--@source(type='kafka',
--        topic.list='localtopic',
--        partition.no='1',
--        threading.option='single.thread',
--        group.id="test",
--        bootstrap.servers='localhost:9092',
--        @map(type='json'))
@source(type = 'http',
        receiver.url='http://localhost:8081/Api/public/test',
        basic.auth.enabled='false',@map(type='json'))
define stream FooStream (symbol string, price float, volume long);

--
--@sink(type='kafka',
--    topic='resulttopic',
--    partition.no='0',
--    bootstrap.servers='localhost:9092',
--    @map(type='json')
--)
@sink(type='log', prefix="LOGGER")
define stream OutputStream (symbol string, price float, volume long,multiple float);
@info(name = 'query1')
from FooStream select symbol, price, volume,(price * volume) as multiple  insert into OutputStream;

@App:name("firstrule")
@App:description("Description of the plan") 

@source(type='http',receiver.url = 'http://localhost:8080/exchange',headers = "'content-type:application/json'",@map(type='json'))
define stream exchangeStream(id int, price double, exchangeWay int, exChangeNum long);
@sink(type='log',@map(type='json'))
define stream outputStream(times string, id int, invalidCount long,price double);
define window dataCacheWindow (times string, id int, invalidCount long,price double) timeBatch(10 second);

partition with (id of exchangeStream)
begin

from exchangeStream select time:currentTime() as times, id, count(id) as invalidCount, price group by id having invalidCount > 1 insert into dataCacheWindow;
from dataCacheWindow select times, id, invalidCount, price insert into outputStream;

end