pipestreams_sdk_quickstart_source_transform_sink_pipeline.ts

1import { PipeStreamsClient, StreamConfig } from '@pipestreams/sdk';
2
3// Initialize the client with your API key
4const client = new PipeStreamsClient({
5  apiKey: 'YOUR_API_KEY_HERE',
6  region: 'us-east-1'
7});
8
9async function main() {
10  try {
11    // 1. Define the stream configuration
12    const config: StreamConfig = {
13      name: 'QuickstartStream',
14      source: {
15        type: 'http',
16        url: 'https://api.example.com/data-feed'
17      },
18      transform: (data) => {
19        // Simple transformation: add a timestamp to each record
20        return {
21          ...data,
22          processedAt: new Date().toISOString()
23        };
24      },
25      sink: {
26        type: 'stdout'
27      }
28    };
29
30    // 2. Create and start the stream
31    const stream = await client.createStream(config);
32    
33    console.log(`Stream started with ID: ${stream.id}`);
34
35    // 3. Listen for lifecycle events
36    stream.on('data', (chunk) => {
37      console.log('Processed chunk:', chunk);
38    });
39
40    stream.on('error', (err) => {
41      console.error('Stream error:', err);
42    });
43
44    stream.on('end', () => {
45      console.log('Stream processing complete.');
46    });
47
48  } catch (error) {
49    console.error('Failed to initialize PipeStream:', error);
50  }
51}
52
53main();