tardieu

package main

import (
	"fmt"
	"net/http"
	"os"
)

const header = "Knative-Serving-Tag"

tardieu

package main

import (
	"fmt"
	"net/http"
	"os"
)

func main() {
	// http request handler

tardieu

source.send_to(lambda event: sink.append(event))

tardieu

sink_config = dict(kind='slack-sink',
                   channel=slack_channel,
                   webhookUrl=slack_webhook)

sink = rayvens.Stream('slack', sink_config=sink_config)

tardieu

source_config = dict(
    kind='http-source',
    url='http://financialmodelingprep.com/api/v3/quote-short/AAPL?apikey=demo',
    period=3000)

source = rayvens.Stream('http', source_config=source_config)

tardieu

import ray
import rayvens

ray.init()
rayvens.init()

tardieu

@ray.remote
class Comparator:
    def __init__(self):
        self.last_quote = None

    def append(self, event):
        payload = json.loads(event)  # parse event payload to json
        quote = payload[0]['price']  # extract AAPL quote
        if self.last_quote:
            if quote > self.last_quote:

tardieu

apiVersion: ibmcloud.ibm.com/v1alpha1
kind: Composable
metadata:
  name: source
spec:
  template:
    apiVersion: sources.eventing.knative.dev/v1alpha1
    kind: KafkaSource
    metadata:
      name: source

tardieu

let { Bundle, EventStreams } = require('solsa')

let bundle = new Bundle()
bundle.kafka = new EventStreams({ name: 'kafka', plan: 'standard' })
bundle.topic = new bundle.kafka.Topic({ name: 'topic', topicName: 'MyTopic' })
bundle.source = new bundle.topic.Source({ name: 'source', sink: { name: 'test-sink' } })
module.exports = bundle

tardieu

apiVersion: ibmcloud.ibm.com/v1alpha1
kind: Topic
metadata:
  name: topic
spec:
  bindingFrom:
    name: kafka
  topicName: MyTopic