Connecteurs, Transforms & SMTs

1package com.example.connector;
2
3import com.streamflow.connect.source.sdk.EventCollector;
4import com.streamflow.connect.source.sdk.SourceConnectorPlugin;
5import java.net.URI;
6import java.net.http.*;
7import java.nio.charset.StandardCharsets;
8import java.util.Map;
9import java.util.concurrent.*;
10
11/**
12 * Source connector that polls a weather API and publishes
13 * readings to a StreamFlow topic.
14 *
15 * Configuration (in connect-source.properties):
16 *   source.meteo.type=weather-api
17 *   source.meteo.topic=weather-readings
18 *   source.meteo.api.url=https://api.meteo.example.com/current
19 *   source.meteo.api.key=${WEATHER_API_KEY}
20 *   source.meteo.cities=Paris,Lyon,Marseille
21 *   source.meteo.poll.interval.seconds=30
22 */
23public class WeatherSourceConnector implements SourceConnectorPlugin {
24
25    private String apiUrl;
26    private String apiKey;
27    private String[] cities;
28    private int pollIntervalSeconds;
29    private String topic;
30
31    private HttpClient httpClient;
32    private ScheduledExecutorService scheduler;
33    private EventCollector collector;
34    private volatile boolean healthy = true;
35
36    // ── Connector identity ──
37
38    @Override
39    public String id() {
40        return "weather-api";  // used in source.<name>.type=weather-api
41    }
42
43    @Override
44    public String name() {
45        return "Weather API Source Connector";
46    }
47
48    // ── Configuration ──
49    // The "source.<name>." prefix is automatically stripped.
50    // You receive: {"api.url": "...", "api.key": "...", "cities": "..."}
51
52    @Override
53    public void configure(Map<String, String> config) {
54        this.apiUrl = config.getOrDefault("api.url",
55            "https://api.meteo.example.com/current");
56        this.apiKey = config.getOrDefault("api.key", "");
57        this.topic = config.getOrDefault("topic", "weather-readings");
58        this.cities = config.getOrDefault("cities", "Paris").split(",");
59        this.pollIntervalSeconds = Integer.parseInt(
60            config.getOrDefault("poll.interval.seconds", "30"));
61    }
62
63    // ── Startup ──
64    // The `collector` is provided by StreamFlow. Call collector.collect()
65    // to send events to the topic.
66
67    @Override
68    public void start(EventCollector collector) {
69        this.collector = collector;
70        this.httpClient = HttpClient.newHttpClient();
71        this.scheduler = Executors.newSingleThreadScheduledExecutor();
72
73        // Poll the API at regular intervals
74        scheduler.scheduleAtFixedRate(
75            this::pollWeather,
76            0,                          // start immediately
77            pollIntervalSeconds,        // then every N seconds
78            TimeUnit.SECONDS
79        );
80
81        System.out.printf("[%s] Started: %d cities, interval %ds%n",
82            name(), cities.length, pollIntervalSeconds);
83    }
84
85    private void pollWeather() {
86        for (String city : cities) {
87            try {
88                // 1. Call the API
89                HttpRequest request = HttpRequest.newBuilder()
90                    .uri(URI.create(apiUrl + "?city=" + city.trim()
91                        + "&key=" + apiKey))
92                    .GET()
93                    .build();
94
95                HttpResponse<String> response = httpClient.send(
96                    request, HttpResponse.BodyHandlers.ofString());
97
98                if (response.statusCode() == 200) {
99                    // 2. Send to StreamFlow
100                    collector.collect(
101                        city.trim(),                                      // key
102                        response.body().getBytes(StandardCharsets.UTF_8),  // payload
103                        Map.of(                                           // headers
104                            "source", "weather-api",
105                            "city", city.trim(),
106                            "timestamp", String.valueOf(
107                                System.currentTimeMillis())
108                        )
109                    );
110                }
111
112            } catch (Exception e) {
113                System.err.printf("[%s] Error for %s: %s%n",
114                    name(), city, e.getMessage());
115                healthy = false;
116            }
117        }
118
119        // 3. Force flush (otherwise events accumulate in the batch)
120        collector.flush();
121    }
122
123    // ── Health check (called by StreamFlow for monitoring) ──
124
125    @Override
126    public boolean isHealthy() {
127        return healthy;
128    }
129
130    // ── Graceful shutdown ──
131
132    @Override
133    public void stop() {
134        if (scheduler != null) {
135            scheduler.shutdown();
136            try {
137                scheduler.awaitTermination(5, TimeUnit.SECONDS);
138            } catch (InterruptedException e) {
139                Thread.currentThread().interrupt();
140            }
141        }
142    }
143
144    @Override
145    public void close() {
146        stop();
147    }
148}

1com.example.connector.WeatherSourceConnector

1<!-- pom.xml -->
2<project>
3    <groupId>com.example</groupId>
4    <artifactId>weather-source-connector</artifactId>
5    <version>1.0.0</version>
6    <packaging>jar</packaging>
7
8    <dependencies>
9        <dependency>
10            <groupId>com.streamflow</groupId>
11            <artifactId>streamflow-connect</artifactId>
12            <version>0.1.0-SNAPSHOT</version>
13            <scope>provided</scope>  <!-- provided by StreamFlow at runtime -->
14        </dependency>
15    </dependencies>
16</project>

1package com.example.connector;
2
3import com.streamflow.connect.connector.*;
4import com.streamflow.connect.config.ConnectorConfig;
5import com.streamflow.connect.sink.*;
6import java.net.URI;
7import java.net.http.*;
8import java.nio.charset.StandardCharsets;
9import java.util.List;
10import java.util.Map;
11
12/**
13 * Sink connector that sends events to a Slack channel.
14 *
15 * Configuration:
16 *   sink.slack.type=slack
17 *   sink.slack.webhook.url=${SLACK_WEBHOOK_URL}
18 *   sink.slack.channel=#alerts
19 *   sink.slack.username=StreamFlow Bot
20 */
21public class SlackSinkConnector implements SinkConnector {
22
23    private String webhookUrl;
24    private String channel;
25    private String username;
26    private HttpClient httpClient;
27    private ConnectorStatus status =
28        new ConnectorStatus(ConnectorStatus.State.STOPPED, null);
29
30    @Override
31    public String connectorId() {
32        return "slack-sink";
33    }
34
35    @Override
36    public ConnectorType type() {
37        return ConnectorType.SINK;
38    }
39
40    @Override
41    public void initialize(ConnectorConfig config) {
42        this.webhookUrl = config.getString("webhook.url");
43        this.channel = config.getString("channel", "#general");
44        this.username = config.getString("username", "StreamFlow");
45    }
46
47    @Override
48    public void start() {
49        this.httpClient = HttpClient.newHttpClient();
50        this.status = new ConnectorStatus(
51            ConnectorStatus.State.RUNNING, null);
52        System.out.printf("[Slack Sink] Started → channel %s%n", channel);
53    }
54
55    // ── Receiving events ──
56    // StreamFlow calls put() with a batch of records
57
58    @Override
59    public void put(List<SinkRecord> records) {
60        for (SinkRecord record : records) {
61            try {
62                String message = new String(
63                    record.value(), StandardCharsets.UTF_8);
64
65                // Build Slack payload
66                String slackPayload = """
67                    {
68                      "channel": "%s",
69                      "username": "%s",
70                      "text": "*[%s]* key=`%s`\n```%s```"
71                    }
72                    """.formatted(channel, username,
73                        record.topic(), record.key(), message);
74
75                // Send via webhook
76                HttpRequest request = HttpRequest.newBuilder()
77                    .uri(URI.create(webhookUrl))
78                    .header("Content-Type", "application/json")
79                    .POST(HttpRequest.BodyPublishers.ofString(slackPayload))
80                    .build();
81
82                httpClient.send(request,
83                    HttpResponse.BodyHandlers.ofString());
84
85            } catch (Exception e) {
86                System.err.printf("[Slack Sink] Error: %s%n",
87                    e.getMessage());
88            }
89        }
90    }
91
92    @Override
93    public void flush(Map<String, SinkOffset> offsets) {
94        // Slack is fire-and-forget, no checkpoint needed
95    }
96
97    @Override
98    public SinkOffset currentOffset(String topic, int partition) {
99        return null;  // no offset resume for Slack
100    }
101
102    @Override
103    public ConnectorStatus status() {
104        return status;
105    }
106
107    @Override
108    public ConnectorMetadata metadata() {
109        return new ConnectorMetadata(
110            "slack-sink", "Slack Sink Connector",
111            "Sends events to a Slack channel via webhook",
112            "1.0.0", "My Company", ConnectorType.SINK,
113            List.of("json"), List.of()
114        );
115    }
116
117    @Override
118    public void stop() {
119        status = new ConnectorStatus(
120            ConnectorStatus.State.STOPPED, null);
121    }
122
123    @Override
124    public void close() {
125        stop();
126    }
127}

1com.example.connector.SlackSinkConnector

1package com.example.transform;
2
3import com.streamflow.connect.source.sdk.MessageTransform;
4import java.time.Instant;
5import java.time.ZoneId;
6import java.time.format.DateTimeFormatter;
7import java.util.*;
8
9/**
10 * Adds a "timestamp-iso" header with the ISO-8601 formatted date.
11 *
12 * Configuration:
13 *   source.my-source.transforms=add-timestamp
14 *   source.my-source.transforms.add-timestamp.type=timestamp-iso
15 *   source.my-source.transforms.add-timestamp.timezone=Europe/Paris
16 */
17public class TimestampIsoTransform implements MessageTransform {
18
19    private DateTimeFormatter formatter;
20
21    @Override
22    public String type() {
23        return "timestamp-iso";
24    }
25
26    @Override
27    public void configure(Map<String, String> config) {
28        String timezone = config.getOrDefault("timezone", "UTC");
29        this.formatter = DateTimeFormatter.ISO_OFFSET_DATE_TIME
30            .withZone(ZoneId.of(timezone));
31    }
32
33    @Override
34    public TransformResult apply(
35            String key, byte[] payload, Map<String, String> headers) {
36        // Add header without modifying the payload
37        Map<String, String> newHeaders = new HashMap<>(headers);
38        newHeaders.put("timestamp-iso", formatter.format(Instant.now()));
39        return new TransformResult(key, payload, newHeaders);
40    }
41
42    @Override
43    public void close() {}
44}

1package com.example.transform;
2
3import com.streamflow.connect.source.sdk.MessageTransform;
4import java.nio.charset.StandardCharsets;
5import java.util.Map;
6
7/**
8 * Filters events whose amount is below the threshold.
9 * Returning null = the event is dropped.
10 *
11 * Configuration:
12 *   source.orders.transforms=min-amount
13 *   source.orders.transforms.min-amount.type=min-amount-filter
14 *   source.orders.transforms.min-amount.field=amount
15 *   source.orders.transforms.min-amount.threshold=100.0
16 */
17public class MinAmountFilter implements MessageTransform {
18
19    private String field;
20    private double threshold;
21
22    @Override
23    public String type() {
24        return "min-amount-filter";
25    }
26
27    @Override
28    public void configure(Map<String, String> config) {
29        this.field = config.getOrDefault("field", "amount");
30        this.threshold = Double.parseDouble(
31            config.getOrDefault("threshold", "0"));
32    }
33
34    @Override
35    public TransformResult apply(
36            String key, byte[] payload, Map<String, String> headers) {
37        String json = new String(payload, StandardCharsets.UTF_8);
38
39        // Simple amount extraction (use Jackson in production)
40        String search = "\"" + field + "\":";
41        int idx = json.indexOf(search);
42        if (idx >= 0) {
43            String numStr = json.substring(
44                idx + search.length()).trim();
45            StringBuilder sb = new StringBuilder();
46            for (char c : numStr.toCharArray()) {
47                if (Character.isDigit(c) || c == '.') sb.append(c);
48                else break;
49            }
50            double amount = Double.parseDouble(sb.toString());
51
52            if (amount < threshold) {
53                return null;  // ← DROP: event is removed
54            }
55        }
56
57        return new TransformResult(key, payload, headers);  // ← KEEP
58    }
59
60    @Override
61    public void close() {}
62}

1source.users.transforms=mask-pii
2source.users.transforms.mask-pii.type=mask-fields
3source.users.transforms.mask-pii.fields=email,phone,ssn
4source.users.transforms.mask-pii.replacement=***

1# connect-source.properties
2
3# Source connector
4source.orders.type=http
5source.orders.topic=orders-clean
6source.orders.url=https://api.shop.com/orders
7
8# Chain of 3 transforms (in order)
9source.orders.transforms=filter-cancelled,add-region,mask-card
10
11# Transform 1: Drop cancelled orders
12source.orders.transforms.filter-cancelled.type=filter
13source.orders.transforms.filter-cancelled.not.contains=CANCELLED
14
15# Transform 2: Add a "region" header
16source.orders.transforms.add-region.type=header-enrich
17source.orders.transforms.add-region.header.region=eu-west
18source.orders.transforms.add-region.header.env=production
19
20# Transform 3: Mask credit card number
21source.orders.transforms.mask-card.type=mask-fields
22source.orders.transforms.mask-card.fields=credit_card,cvv
23source.orders.transforms.mask-card.replacement=****

1# connect-source.properties
2
3# ── Pattern: source.<name>.<property> ──
4
5# MQTT
6source.iot-sensors.type=mqtt
7source.iot-sensors.topic=sensor-data
8source.iot-sensors.broker=tcp://mqtt.prod.internal:1883
9source.iot-sensors.subscribe=sensors/#
10source.iot-sensors.qos=1
11source.iot-sensors.username=${MQTT_USER}
12source.iot-sensors.password=${MQTT_PASSWORD}
13
14# PostgreSQL CDC (change data capture)
15source.orders-cdc.type=cdc
16source.orders-cdc.topic=cdc-orders
17source.orders-cdc.database.type=postgresql
18source.orders-cdc.database.hostname=db.prod.internal
19source.orders-cdc.database.port=5432
20source.orders-cdc.database.name=shop
21source.orders-cdc.database.user=cdc_user
22source.orders-cdc.database.password=${DB_PASSWORD}
23source.orders-cdc.tables=public.orders,public.order_items
24source.orders-cdc.poll.interval.ms=1000
25
26# HTTP webhook
27source.payments.type=http
28source.payments.topic=payment-events
29source.payments.url=https://payments.example.com/webhook
30source.payments.method=GET
31source.payments.poll.interval.ms=5000
32source.payments.headers.Authorization=Bearer ${PAYMENT_API_KEY}
33
34# AMQP (RabbitMQ)
35source.notifications.type=amqp
36source.notifications.topic=notifications
37source.notifications.host=rabbitmq.internal
38source.notifications.port=5672
39source.notifications.queue=notification-queue
40source.notifications.username=${RABBIT_USER}
41source.notifications.password=${RABBIT_PASSWORD}

1export MQTT_USER=admin
2export MQTT_PASSWORD=secret123
3export DB_PASSWORD=pg_secure_pass
4export PAYMENT_API_KEY=pk_live_abc123

1# 1. Build the connector
2cd weather-source-connector/
3mvn package
4
5# 2. Copy to StreamFlow plugins directory
6cp target/weather-source-connector-1.0.0.jar /opt/streamflow/connectors/
7
8# 3. Configure
9cat >> /opt/streamflow/conf/connect-source.properties << 'EOF'
10source.meteo-paris.type=weather-api
11source.meteo-paris.topic=weather
12source.meteo-paris.api.url=https://api.meteo.example.com/current
13source.meteo-paris.api.key=${WEATHER_API_KEY}
14source.meteo-paris.cities=Paris,Lyon,Marseille
15source.meteo-paris.poll.interval.seconds=60
16EOF
17
18# 4. Restart StreamFlow (connector is auto-discovered)
19pkill -f StreamFlowApplication
20./scripts/start-streamflow.sh

1# Dockerfile
2FROM streamflow/streamflow:latest
3
4# Add custom connector
5COPY weather-source-connector-1.0.0.jar /opt/streamflow/connectors/
6
7# Add config
8COPY connect-source.properties /opt/streamflow/conf/connect-source.properties

1# docker-compose.yml
2services:
3  streamflow:
4    build: .
5    ports:
6      - "9092:9092"
7      - "8080:8080"
8    environment:
9      WEATHER_API_KEY: "my-api-key"