suhothayan

@App:name('NotificationManagement')
@App:description('Streaming App for Notification Management')

@source(type = 'kafka',
		topic.list = 'shipment-info', bootstrap.servers = 'localhost:9092', 
		group.id = 'test', threading.option = 'single.thread', 
		is.binary.message = 'true',
	@map(type = 'avro'))
define stream ShipmentInfoStream (orderId string, buyerId string, shipmentType string);

suhothayan

package io.siddhi.sample;

import io.siddhi.core.SiddhiAppRuntime;
import io.siddhi.core.SiddhiManager;
import io.siddhi.core.event.Event;
import io.siddhi.core.stream.input.InputHandler;
import io.siddhi.core.stream.output.StreamCallback;
import io.siddhi.core.util.EventPrinter;

/**

suhothayan

@App:name('CountOverTime')
@App:description('Receive events via HTTP, and logs the number of events received during last 15 seconds')

@source(type = 'http', receiver.url = "http://0.0.0.0:8006/production",
	@map(type = 'json'))
define stream ProductionStream (name string, amount double);

@sink(type = 'log')
define stream TotalCountStream (totalCount long);

suhothayan

@App:name("ConsumeAndStore")
@App:description("Consume events from HTTP and write to TEST_DB")

@source(type = 'http', receiver.url = "http://0.0.0.0:8006/production",
	@map(type = 'json'))
define stream ProductionStream (name string, amount double);

@store(type='rdbms', datasource='TEST_DB')
define table ProductionTable (name string, amount double);

suhothayan

wso2.datasources:
  dataSources:
  - name: TEST_DB
    description: The datasource used for testing
    definition:
      type: RDBMS
      configuration:
        jdbcUrl: 'jdbc:h2:${sys:carbon.home}/wso2/${sys:wso2.runtime}/database/TEST_DB;DB_CLOSE_ON_EXIT=FALSE;LOCK_TIMEOUT=60000'
        username: admin
        password: admin

suhothayan

@App:name("TemplatedFilterAndEmail")
@App:description("Consumes events from HTTP, filters them based on amount greater than a templated threshold value, and sends filtered events via email.")

@source(type = 'http', receiver.url = "http://0.0.0.0:8006/production",
	@map(type = 'json'))
define stream ProductionStream (name string, amount double);

@sink(ref = 'email-sink', subject = 'High {{name}} production!', to = '${TO_EMAIL}', content.type = 'text/html',
	@map(type = 'text',
		@payload("""

suhothayan

siddhi:
  refs:
    -
      ref:
        name: 'email-sink'
        type: 'email'
        properties:
          port: '465'
          host: 'smtp.gmail.com'
          ssl.enable: 'true'