add coinbase sample

examples/coinbase/Makefile

@@ -0,0 +1,10 @@
+
+
+PWD = $(shell pwd)
+NAME = test
+
+create:	
+	yq -o=json $(PWD)/pipeline.yaml | curl http://localhost:4195/streams/$(NAME) -X POST  -d  @- 
+
+delete:
+	curl http://localhost:4195/streams/$(NAME) -X DELETE

examples/coinbase/README.md

@@ -0,0 +1,89 @@
+# Demo for Benthos data pipeline and Coinbase websocket data
+
+
+
+This docker compose file demonstrates how to ingest WebSocket data into Proton by using Benthos pipeline. 
+
+
+
+## Start the stack
+
+Simply run `docker compose up` in this folder. Three docker containers in the stack:
+
+1. ghcr.io/timeplus-io/proton:latest, as the streaming database
+2. jeffail/benthos:latest, a [Benthos](https://www.benthos.dev/) service as the data pipeline
+3. init container, create the tickers stream when Proton database server is ready
+
+the ddl to create the stream is:
+
+```sql
+CREATE STREAM IF NOT EXISTS tickers (
+    best_ask decimal(10,2),
+    best_ask_size decimal(10,8),
+    best_bid decimal(10,2),
+    best_bid_size decimal(10,8),
+    high_24h decimal(10,2),
+    last_size decimal(10,8),
+    low_24h decimal(10,2),
+    open_24h decimal(10,2),
+    price decimal(10,2),
+    product_id string,
+    sequence int,
+    side string,
+    time datetime,
+    trade_id int,
+    type string,
+    volume_24h decimal(20,8),
+    volume_30d decimal(20,8)
+)
+```
+
+## Create a ingest data pipeline
+
+run command `make create` to create following Benthos data pipeline, note you need install `jq` and `curl` to run this command
+
+```
+input:
+  label: coinbase
+  websocket:
+    url:  wss://ws-feed.exchange.coinbase.com
+    open_message: '{"type": "subscribe","product_ids": ["ETH-USD","ETH-EUR"],"channels": ["ticker"]}'
+    open_message_type: text
+
+output:
+  http_client:
+    url: http://proton:8123/proton/v1/ingest/streams/tickers
+    verb: POST
+    headers: 
+      Content-Type: application/json
+    batching:
+      count: 10
+      period: 1000ms
+      processors:
+        - archive:
+            format: json_array
+        - mapping: |
+            root.columns = this.index(0).keys()
+            root.data = this.map_each( row -> root.columns.map_each( key -> row.get(key)) )
+
+```
+
+this pipeline will read data from coinbase websocket and then send the result to proton ingest api in a batch
+
+
+## Query you crypto price data with SQL
+
+now you can run following query to get the OHLC of the crypto data:
+
+```sql
+SELECT
+  window_start, product_id, earliest(price) AS open, max(price) AS high, min(price) AS low, latest(price) AS close
+FROM
+  tumble(tickers, 60s)
+WHERE
+  product_id != '' and _tp_time > earliest_ts()
+GROUP BY
+  window_start, product_id
+```
+
+

examples/coinbase/ddl.sql

@@ -0,0 +1,21 @@
+--  ddl creare stream
+CREATE STREAM IF NOT EXISTS tickers (
+    best_ask decimal(10,2),
+    best_ask_size decimal(10,8),
+    best_bid decimal(10,2),
+    best_bid_size decimal(10,8),
+    high_24h decimal(10,2),
+    last_size decimal(10,8),
+    low_24h decimal(10,2),
+    open_24h decimal(10,2),
+    price decimal(10,2),
+    product_id string,
+    sequence int,
+    side string,
+    time datetime,
+    trade_id int,
+    type string,
+    volume_24h decimal(20,8),
+    volume_30d decimal(20,8)
+)
+

examples/coinbase/docker-compose.yml

@@ -0,0 +1,34 @@
+version: '3.7'
+name: coinbase
+services:
+  proton:
+    image: ghcr.io/timeplus-io/proton:latest
+    pull_policy: always
+    ports:
+    - "3218:3218" # HTTP Streaming
+    healthcheck:
+      test: ["CMD", "curl", "http://localhost:3218/proton/ping"]
+      interval: 2s
+      timeout: 10s
+      retries: 3
+      start_period: 10s
+
+  benthos:
+    image: jeffail/benthos:latest
+    pull_policy: always
+    command: streams
+    ports:
+    - "4195:4195"
+
+  init-stream:
+    image: ghcr.io/timeplus-io/proton:latest
+    command:
+      - sh
+      - -c
+      - |
+        proton-client -h proton --query "CREATE STREAM IF NOT EXISTS tickers (best_ask decimal(10,2), best_ask_size decimal(10,8), best_bid decimal(10,2), best_bid_size decimal(10,8), high_24h decimal(10,2), last_size decimal(10,8), low_24h decimal(10,2), open_24h decimal(10,2), price decimal(10,2), product_id string, sequence int, side string, time datetime, trade_id int, type string, volume_24h decimal(20,8), volume_30d decimal(20,8))"
+    depends_on:
+      proton:
+        condition: service_healthy
+
+

examples/coinbase/pipeline.yaml

@@ -0,0 +1,22 @@
+input:
+  label: coinbase
+  websocket:
+    url:  wss://ws-feed.exchange.coinbase.com
+    open_message: '{"type": "subscribe","product_ids": ["ETH-USD","ETH-EUR"],"channels": ["ticker"]}'
+    open_message_type: text
+
+output:
+  http_client:
+    url: http://proton:8123/proton/v1/ingest/streams/tickers
+    verb: POST
+    headers: 
+      Content-Type: application/json
+    batching:
+      count: 10
+      period: 1000ms
+      processors:
+        - archive:
+            format: json_array
+        - mapping: |
+            root.columns = this.index(0).keys()
+            root.data = this.map_each( row -> root.columns.map_each( key -> row.get(key)) )