apiVersion: 0.1.0
meta:
  version: 0.2.4
  name: http-github
  type: http-source
  topic: github-events
  secrets:
    - name: GITHUB_TOKEN
http:
  endpoint: 'https://api.github.com/repos/infinyon/fluvio'
  method: GET
  headers: 
    - 'Authorization: token ${{ secrets.GITHUB_TOKEN }}'
  interval: 30s

apiVersion: 0.1.0
meta:
  version: 0.2.4
  name: helsinki-mqtt 
  type: mqtt-source
  topic: helsinki
mqtt:
  url: "mqtt://mqtt.hsl.fi"
  topic: "/hfp/v2/journey/ongoing/vp/+/+/+/#"
  timeout:
    secs: 30
    nanos: 0
  payload_output_type: json

use fluvio::{Fluvio, RecordKey};

#[async_std::main]
async fn main() {
   let topic = "rust-topic";
   let record = "Hello from rust!";

   let fluvio = Fluvio::connect().await.unwrap();

   let producer = fluvio::producer(topic).await.unwrap();
   producer.send(RecordKey::NULL, record).await.unwrap();

   producer.flush().await.unwrap();
}

#!/usr/bin/env python
from fluvio import Fluvio

if __name__ == "__main__":
    topic = "python-topic";
    record = "Hello from rust!";

    fluvio = Fluvio.connect()

    producer = fluvio.topic_producer(topic)
    producer.send_string(record)

    producer.flush()

import Fluvio from "@fluvio/client";

const produce = async () => {
   const TOPIC_NAME = "node-topic";
   const RECORD_TEXT = "Hello from node!"

   await fluvio.connect();

   const producer = await fluvio.topicProducer(TOPIC_NAME);
   await producer.send(RECORD_TEXT);
};

const fluvio = new Fluvio();
produce();

apiVersion: 0.1.0
meta:
  version: 0.2.4
  name: my-http-sink
  type: http-sink
  topic: http-sink-topic
  secrets:
    - name: AUTHORIZATION_TOKEN
http:
  endpoint: "http://127.0.0.1/post"
  headers:
    - "Authorization: token ${{ secrets.AUTHORIZATION_TOKEN }}"
    - "Content-Type: application/json"

apiVersion: 0.1.0
meta:
  version: 0.3.1
  name: json-sql-connector
  type: sql-sink
  topic: sql-topic
  secrets:
    - name: DATABASE_URL
sql:
  url: ${{ secrets.DATABASE_URL }}
transforms:
  - uses: infinyon/json-sql
    with:
      mapping:
        table: "topic_message"
        map-columns:
          "device_id":
            json-key: "device.device_id"
            value:
              type: "int"
              default: "0"
              required: true
          "record":
            json-key: "$"
            value:
              type: "jsonb"
              required: true

apiVersion: 0.1.0
meta:
  version: 0.1.1
  name: weather-monitor-sandiego
  type: graphite-sink
  topic: weather-ca-sandiego
graphite:
  # https://graphite.readthedocs.io/en/latest/feeding-carbon.html#step-1-plan-a-naming-hierarchy
  metric-path: "weather.temperature.ca.sandiego"
  addr: "localhost:2003"

apiVersion: 0.1.0
meta:
  version: 0.1.0
  name: duckdb-connector
  type: duckdb-sink
  topic: fluvio-topic-source
duckdb:
  url: 'local.db'    # local duckdb

use async_std::stream::StreamExt;
use fluvio::{Fluvio, Offset};

#[async_std::main]
async fn main() {
   let topic = "rust-topic";
   let partition = 0;

   let fluvio = Fluvio::connect().await.unwrap();

   let consumer = fluvio::consumer(topic, partitino).await.unwrap();
   let mut stream = consumer.stream(Offset::from_end(1)).await.unwrap();
   if let Some(Ok(record)) = stream.next().await {
       let string = String::from_utf8_lossy(record.value());
       println!("{}", string);
   }
}

#!/usr/bin/env python
from fluvio import Fluvio, Offset

if __name__ == "__main__":
    topic = "python-topic";
    partition = 0;

    fluvio = Fluvio.connect()

    consumer = fluvio.partition_consumer(topic, partition)
    for idx, record in enumerate( consumer.stream(Offset.from_end(10)) ):
       print("{}".format(record.value_string()))
       if idx >= 9:
           break

import Fluvio, { Offset, Record } from "@fluvio/client";

const consume = async () => {
   const TOPIC_NAME = "hello-node";
   const PARTITION = 0;

   await fluvio.connect();

   const consumer = await fluvio.partitionConsumer(TOPIC_NAME, PARTITION);
   await consumer.stream(Offset.FromEnd(), async (record: Record) => {
      console.log(`Key=${record.keyString()}, Value=${record.valueString()}`);
      process.exit(0);
   });
};

const fluvio = new Fluvio();
consume();