1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68

syntax = "proto3";
 
package com.example.streaming.v1;
 
// Complete service demonstrating all four patterns
service DataService {
  // Pattern 1: Unary RPC
  // Client sends one request, receives one response
  rpc GetItem(GetItemRequest) returns (GetItemResponse);
  
  // Pattern 2: Server Streaming RPC
  // Client sends one request, receives stream of responses
  rpc ListItems(ListItemsRequest) returns (stream Item);
  
  // Pattern 3: Client Streaming RPC
  // Client sends stream of requests, receives one response
  rpc UploadItems(stream Item) returns (UploadSummary);
  
  // Pattern 4: Bidirectional Streaming RPC
  // Both sides send streams independently
  rpc ProcessItems(stream ItemOperation) returns (stream OperationResult);
}
 
message GetItemRequest {
  string item_id = 1;
}
 
message GetItemResponse {
  Item item = 1;
}
 
message ListItemsRequest {
  string filter = 1;
  int32 page_size = 2;
}
 
message Item {
  string id = 1;
  string name = 2;
  bytes data = 3;
  int64 timestamp = 4;
}
 
message UploadSummary {
  int32 items_received = 1;
  int32 items_processed = 2;
  int32 items_failed = 3;
  repeated string error_ids = 4;
}
 
message ItemOperation {
  string item_id = 1;
  OperationType type = 2;
  bytes payload = 3;
}
 
message OperationResult {
  string item_id = 1;
  bool success = 2;
  string error_message = 3;
}
 
enum OperationType {
  OPERATION_TYPE_UNSPECIFIED = 0;
  OPERATION_TYPE_CREATE = 1;
  OPERATION_TYPE_UPDATE = 2;
  OPERATION_TYPE_DELETE = 3;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90

// Unary RPC: Client Implementation
 
import { DataServiceClient } from './generated/data_service';
import { credentials, Metadata, CallOptions } from '@grpc/grpc-js';
 
const client = new DataServiceClient(
    'localhost:50051',
    credentials.createInsecure()
);
 
// Simple unary call
async function getItem(itemId: string): Promise<Item> {
    return new Promise((resolve, reject) => {
        client.getItem(
            { itemId },
            (error, response) => {
                if (error) {
                    reject(error);
                } else {
                    resolve(response.item);
                }
            }
        );
    });
}
 
// Unary call with metadata and deadline
async function getItemWithOptions(itemId: string): Promise<Item> {
    const metadata = new Metadata();
    metadata.set('authorization', 'Bearer <token>');
    metadata.set('x-request-id', crypto.randomUUID());
    
    const deadline = new Date();
    deadline.setSeconds(deadline.getSeconds() + 5);  // 5 second timeout
    
    const options: CallOptions = {
        deadline,
    };
    
    return new Promise((resolve, reject) => {
        const call = client.getItem(
            { itemId },
            metadata,
            options,
            (error, response) => {
                if (error) {
                    console.error(`RPC failed: ${error.code} - ${error.message}`);
                    reject(error);
                } else {
                    resolve(response.item);
                }
            }
        );
        
        // Access response metadata
        call.on('metadata', (metadata: Metadata) => {
            console.log('Response headers:', metadata.getMap());
        });
        
        call.on('status', (status) => {
            console.log('Trailing metadata:', status.metadata.getMap());
        });
    });
}
 
// Unary call with cancellation
function getItemCancellable(itemId: string): { 
    promise: Promise<Item>; 
    cancel: () => void 
} {
    let call: any;
    
    const promise = new Promise<Item>((resolve, reject) => {
        call = client.getItem(
            { itemId },
            (error, response) => {
                if (error) {
                    reject(error);
                } else {
                    resolve(response.item);
                }
            }
        );
    });
    
    return {
        promise,
        cancel: () => call.cancel(),
    };
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

// Unary RPC: Server Implementation
 
import { sendUnaryData, ServerUnaryCall, status } from '@grpc/grpc-js';
import { GetItemRequest, GetItemResponse, Item } from './generated/data_service';
 
class DataServiceImpl {
    async getItem(
        call: ServerUnaryCall<GetItemRequest, GetItemResponse>,
        callback: sendUnaryData<GetItemResponse>
    ): Promise<void> {
        // Extract request
        const { itemId } = call.request;
        
        // Access client metadata
        const metadata = call.metadata;
        const authHeader = metadata.get('authorization')[0];
        const requestId = metadata.get('x-request-id')[0];
        
        console.log(`Request ${requestId} for item ${itemId}`);
        
        // Check if client already cancelled
        if (call.cancelled) {
            console.log(`Request ${requestId} was cancelled`);
            return;
        }
        
        try {
            // Business logic
            const item = await this.fetchItemFromDatabase(itemId);
            
            if (!item) {
                // Return gRPC NOT_FOUND error
                callback({
                    code: status.NOT_FOUND,
                    message: `Item ${itemId} not found`,
                    details: 'Item does not exist or has been deleted',
                }, null);
                return;
            }
            
            // Set response metadata
            const responseMetadata = new Metadata();
            responseMetadata.set('x-cache-status', 'HIT');
            call.sendMetadata(responseMetadata);
            
            // Return successful response
            callback(null, { item });
            
        } catch (error) {
            // Return gRPC INTERNAL error
            callback({
                code: status.INTERNAL,
                message: 'Internal server error',
                details: error.message,
            }, null);
        }
    }
    
    private async fetchItemFromDatabase(itemId: string): Promise<Item | null> {
        // Database lookup...
        return null;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89

// Server Streaming: Client Implementation
 
import { DataServiceClient } from './generated/data_service';
import { Metadata, ClientReadableStream } from '@grpc/grpc-js';
 
// Basic server streaming consumption
async function listItems(filter: string): Promise<Item[]> {
    const items: Item[] = [];
    
    return new Promise((resolve, reject) => {
        const stream: ClientReadableStream<Item> = client.listItems({
            filter,
            pageSize: 100,
        });
        
        stream.on('data', (item: Item) => {
            items.push(item);
            console.log(`Received item ${item.id}`);
        });
        
        stream.on('end', () => {
            console.log(`Stream ended, received ${items.length} items`);
            resolve(items);
        });
        
        stream.on('error', (error) => {
            console.error(`Stream error: ${error.message}`);
            reject(error);
        });
        
        stream.on('status', (status) => {
            console.log(`gRPC status: ${status.code}`);
        });
    });
}
 
// Server streaming with async iteration (modern approach)
async function* listItemsGenerator(filter: string): AsyncGenerator<Item> {
    const stream = client.listItems({ filter, pageSize: 100 });
    
    for await (const item of stream) {
        yield item;
    }
}
 
// Usage with async iteration
async function processItems(filter: string): Promise<void> {
    for await (const item of listItemsGenerator(filter)) {
        await processItem(item);
        // Backpressure handled automatically
    }
}
 
// Server streaming with cancellation
function listItemsWithTimeout(filter: string, timeoutMs: number): {
    generator: AsyncGenerator<Item>;
    cancel: () => void;
} {
    const stream = client.listItems({ filter, pageSize: 100 });
    let cancelled = false;
    
    const timeoutId = setTimeout(() => {
        stream.cancel();
        cancelled = true;
    }, timeoutMs);
    
    async function* generator(): AsyncGenerator<Item> {
        try {
            for await (const item of stream) {
                clearTimeout(timeoutId);  // Reset on each item
                yield item;
            }
        } catch (error) {
            if (cancelled && error.code === status.CANCELLED) {
                console.log('Stream cancelled due to timeout');
                return;
            }
            throw error;
        }
    }
    
    return {
        generator: generator(),
        cancel: () => {
            stream.cancel();
            cancelled = true;
        },
    };
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98

// Server Streaming: Server Implementation
 
import { ServerWritableStream, status } from '@grpc/grpc-js';
import { ListItemsRequest, Item } from './generated/data_service';
 
class DataServiceImpl {
    async listItems(
        call: ServerWritableStream<ListItemsRequest, Item>
    ): Promise<void> {
        const { filter, pageSize } = call.request;
        
        console.log(`Streaming items with filter: ${filter}`);
        
        try {
            // Database cursor for efficient large result sets
            const cursor = await this.createDatabaseCursor(filter);
            let count = 0;
            
            while (true) {
                // Check for client cancellation
                if (call.cancelled) {
                    console.log('Client cancelled stream');
                    cursor.close();
                    return;
                }
                
                // Fetch next batch
                const batch = await cursor.next(pageSize);
                
                if (batch.length === 0) {
                    break;  // No more items
                }
                
                for (const item of batch) {
                    // Write each item to stream
                    // This respects flow control automatically
                    const canContinue = call.write(item);
                    
                    if (!canContinue) {
                        // Client's buffer is full, wait for drain
                        await new Promise<void>(resolve => {
                            call.once('drain', resolve);
                        });
                    }
                    
                    count++;
                }
            }
            
            console.log(`Streamed ${count} items`);
            
            // End stream successfully
            call.end();
            
        } catch (error) {
            // Send error and end stream
            call.destroy({
                code: status.INTERNAL,
                message: `Stream failed: ${error.message}`,
            });
        }
    }
    
    // Streaming with periodic heartbeats (for long-running streams)
    async streamWithHeartbeat(
        call: ServerWritableStream<SubscribeRequest, Event>
    ): Promise<void> {
        const heartbeatInterval = setInterval(() => {
            if (!call.cancelled) {
                call.write({ type: 'HEARTBEAT', timestamp: Date.now() });
            }
        }, 30000);  // Every 30 seconds
        
        try {
            // Subscribe to event source
            const subscription = await this.eventSource.subscribe(
                call.request.topic,
                (event) => {
                    if (!call.cancelled) {
                        call.write(event);
                    }
                }
            );
            
            // Wait for cancellation
            await new Promise<void>(resolve => {
                call.on('cancelled', resolve);
                call.on('close', resolve);
            });
            
            subscription.unsubscribe();
            
        } finally {
            clearInterval(heartbeatInterval);
            call.end();
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135

// Client Streaming: Client Implementation
 
import { DataServiceClient } from './generated/data_service';
import { ClientWritableStream } from '@grpc/grpc-js';
import { Item, UploadSummary } from './generated/data_service';
import * as fs from 'fs';
 
// Basic client streaming: upload items
async function uploadItems(items: Item[]): Promise<UploadSummary> {
    return new Promise((resolve, reject) => {
        const call: ClientWritableStream<Item> = client.uploadItems(
            (error, response) => {
                if (error) {
                    reject(error);
                } else {
                    resolve(response);
                }
            }
        );
        
        // Write all items
        for (const item of items) {
            call.write(item);
        }
        
        // Signal end of stream
        call.end();
    });
}
 
// Client streaming with backpressure handling
async function uploadItemsWithBackpressure(items: Item[]): Promise<UploadSummary> {
    return new Promise((resolve, reject) => {
        const call: ClientWritableStream<Item> = client.uploadItems(
            (error, response) => {
                if (error) {
                    reject(error);
                } else {
                    resolve(response);
                }
            }
        );
        
        let index = 0;
        
        function writeNext(): void {
            let ok = true;
            
            while (ok && index < items.length) {
                const item = items[index++];
                
                if (index === items.length) {
                    // Last item
                    call.end(item);
                } else {
                    // More items to come
                    ok = call.write(item);
                }
            }
            
            if (index < items.length) {
                // Buffer was full, wait for drain
                call.once('drain', writeNext);
            }
        }
        
        writeNext();
    });
}
 
// File upload with client streaming
async function uploadFile(filePath: string): Promise<UploadSummary> {
    const CHUNK_SIZE = 64 * 1024;  // 64 KB chunks
    
    return new Promise((resolve, reject) => {
        const call = client.uploadItems((error, response) => {
            if (error) reject(error);
            else resolve(response);
        });
        
        const readStream = fs.createReadStream(filePath, {
            highWaterMark: CHUNK_SIZE,
        });
        
        readStream.on('data', (chunk: Buffer) => {
            const item: Item = {
                id: filePath,
                name: 'file_chunk',
                data: chunk,
                timestamp: Date.now(),
            };
            
            if (!call.write(item)) {
                // Pause file reading until gRPC buffer drains
                readStream.pause();
                call.once('drain', () => readStream.resume());
            }
        });
        
        readStream.on('end', () => {
            call.end();
        });
        
        readStream.on('error', (error) => {
            call.destroy(error);
            reject(error);
        });
    });
}
 
// Async generator approach
async function uploadFromGenerator(
    items: AsyncGenerator<Item>
): Promise<UploadSummary> {
    return new Promise(async (resolve, reject) => {
        const call = client.uploadItems((error, response) => {
            if (error) reject(error);
            else resolve(response);
        });
        
        try {
            for await (const item of items) {
                // Wrap write in promise for backpressure
                const canContinue = call.write(item);
                if (!canContinue) {
                    await new Promise<void>(r => call.once('drain', r));
                }
            }
            call.end();
        } catch (error) {
            call.destroy(error as Error);
            reject(error);
        }
    });
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98

// Client Streaming: Server Implementation
 
import { ServerReadableStream, sendUnaryData, status } from '@grpc/grpc-js';
import { Item, UploadSummary } from './generated/data_service';
 
class DataServiceImpl {
    async uploadItems(
        call: ServerReadableStream<Item, UploadSummary>,
        callback: sendUnaryData<UploadSummary>
    ): Promise<void> {
        const results = {
            itemsReceived: 0,
            itemsProcessed: 0,
            itemsFailed: 0,
            errorIds: [] as string[],
        };
        
        call.on('data', async (item: Item) => {
            results.itemsReceived++;
            
            try {
                await this.processItem(item);
                results.itemsProcessed++;
            } catch (error) {
                results.itemsFailed++;
                results.errorIds.push(item.id);
                console.error(`Failed to process ${item.id}: ${error.message}`);
            }
        });
        
        call.on('end', () => {
            console.log(`Upload complete: ${results.itemsReceived} received`);
            
            callback(null, {
                itemsReceived: results.itemsReceived,
                itemsProcessed: results.itemsProcessed,
                itemsFailed: results.itemsFailed,
                errorIds: results.errorIds,
            });
        });
        
        call.on('error', (error) => {
            console.error(`Stream error: ${error.message}`);
            callback({
                code: status.INTERNAL,
                message: error.message,
            }, null);
        });
        
        call.on('cancelled', () => {
            console.log('Client cancelled upload');
            // Clean up any partial state
        });
    }
    
    // Alternative: async iterator approach
    async uploadItemsIterator(
        call: ServerReadableStream<Item, UploadSummary>,
        callback: sendUnaryData<UploadSummary>
    ): Promise<void> {
        let received = 0;
        let processed = 0;
        let failed = 0;
        const errorIds: string[] = [];
        
        try {
            for await (const item of call) {
                received++;
                
                try {
                    await this.processItem(item);
                    processed++;
                } catch (error) {
                    failed++;
                    errorIds.push(item.id);
                }
            }
            
            callback(null, {
                itemsReceived: received,
                itemsProcessed: processed,
                itemsFailed: failed,
                errorIds,
            });
            
        } catch (error) {
            callback({
                code: status.INTERNAL,
                message: error.message,
            }, null);
        }
    }
    
    private async processItem(item: Item): Promise<void> {
        // Validate and store item
        await this.database.insert(item);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168

// Bidirectional Streaming: Complete Implementation
 
// Proto definition:
// service ChatService {
//   rpc Chat(stream ChatMessage) returns (stream ChatMessage);
// }
 
// ============ CLIENT SIDE ============
 
import { ClientDuplexStream } from '@grpc/grpc-js';
import { ChatMessage } from './generated/chat_service';
 
class ChatClient {
    private stream: ClientDuplexStream<ChatMessage, ChatMessage> | null = null;
    private messageHandlers: ((msg: ChatMessage) => void)[] = [];
    
    connect(userId: string): void {
        const metadata = new Metadata();
        metadata.set('user-id', userId);
        
        this.stream = client.chat(metadata);
        
        // Handle incoming messages (server → client)
        this.stream.on('data', (message: ChatMessage) => {
            this.messageHandlers.forEach(handler => handler(message));
        });
        
        this.stream.on('end', () => {
            console.log('Server closed the stream');
            this.stream = null;
        });
        
        this.stream.on('error', (error) => {
            console.error(`Stream error: ${error.message}`);
            this.stream = null;
        });
    }
    
    // Send message (client → server)
    send(content: string, recipientId?: string): void {
        if (!this.stream) {
            throw new Error('Not connected');
        }
        
        const message: ChatMessage = {
            id: crypto.randomUUID(),
            content,
            timestamp: Date.now(),
            recipientId: recipientId || '',
        };
        
        this.stream.write(message);
    }
    
    // Register message handler
    onMessage(handler: (msg: ChatMessage) => void): void {
        this.messageHandlers.push(handler);
    }
    
    // Close client's sending stream (can still receive)
    halfClose(): void {
        this.stream?.end();
    }
    
    // Fully disconnect
    disconnect(): void {
        this.stream?.cancel();
        this.stream = null;
    }
}
 
// Usage
const chatClient = new ChatClient();
chatClient.onMessage((msg) => {
    console.log(`[${msg.senderId}]: ${msg.content}`);
});
chatClient.connect('user-123');
chatClient.send('Hello, world!');
 
// ============ SERVER SIDE ============
 
import { ServerDuplexStream, status } from '@grpc/grpc-js';
 
class ChatServiceImpl {
    private connectedUsers = new Map<string, ServerDuplexStream<ChatMessage, ChatMessage>>();
    
    chat(call: ServerDuplexStream<ChatMessage, ChatMessage>): void {
        const userId = call.metadata.get('user-id')[0] as string;
        
        if (!userId) {
            call.destroy({
                code: status.UNAUTHENTICATED,
                message: 'user-id metadata required',
            });
            return;
        }
        
        console.log(`User ${userId} connected`);
        this.connectedUsers.set(userId, call);
        
        // Notify presence
        this.broadcast({
            type: 'PRESENCE',
            content: `${userId} joined`,
            senderId: 'system',
            timestamp: Date.now(),
        }, userId);
        
        // Handle incoming messages from this client
        call.on('data', (message: ChatMessage) => {
            console.log(`Message from ${userId}: ${message.content}`);
            
            if (message.recipientId) {
                // Direct message
                this.sendTo(message.recipientId, {
                    ...message,
                    senderId: userId,
                });
            } else {
                // Broadcast
                this.broadcast({
                    ...message,
                    senderId: userId,
                }, userId);
            }
        });
        
        call.on('end', () => {
            console.log(`User ${userId} stream ended`);
            this.cleanupUser(userId);
        });
        
        call.on('cancelled', () => {
            console.log(`User ${userId} cancelled`);
            this.cleanupUser(userId);
        });
        
        call.on('error', (error) => {
            console.error(`User ${userId} error: ${error.message}`);
            this.cleanupUser(userId);
        });
    }
    
    private sendTo(userId: string, message: ChatMessage): void {
        const stream = this.connectedUsers.get(userId);
        if (stream && !stream.cancelled) {
            stream.write(message);
        }
    }
    
    private broadcast(message: ChatMessage, excludeUserId?: string): void {
        for (const [userId, stream] of this.connectedUsers) {
            if (userId !== excludeUserId && !stream.cancelled) {
                stream.write(message);
            }
        }
    }
    
    private cleanupUser(userId: string): void {
        this.connectedUsers.delete(userId);
        this.broadcast({
            type: 'PRESENCE',
            content: `${userId} left`,
            senderId: 'system',
            timestamp: Date.now(),
        });
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130

// Proper Backpressure Handling
 
// WRONG: Ignoring backpressure
async function badProducer(call: ServerWritableStream<Data>): Promise<void> {
    for (const item of hugeDataset) {
        call.write(item);  // Buffers unbounded, memory explodes
    }
    call.end();
}
 
// CORRECT: Respecting backpressure
async function goodProducer(call: ServerWritableStream<Data>): Promise<void> {
    for (const item of hugeDataset) {
        if (call.cancelled) {
            return;  // Don't produce for cancelled consumers
        }
        
        const canContinue = call.write(item);
        
        if (!canContinue) {
            // Buffer is full, wait for consumer to catch up
            await new Promise<void>(resolve => {
                call.once('drain', resolve);
            });
        }
    }
    call.end();
}
 
// ADVANCED: Rate limiting with token bucket
class RateLimitedProducer {
    private tokens: number;
    private readonly maxTokens: number;
    private readonly refillRate: number;  // tokens per second
    private lastRefill: number;
    
    constructor(maxTokens: number, refillRate: number) {
        this.tokens = maxTokens;
        this.maxTokens = maxTokens;
        this.refillRate = refillRate;
        this.lastRefill = Date.now();
    }
    
    async acquire(): Promise<void> {
        this.refill();
        
        while (this.tokens < 1) {
            await new Promise(r => setTimeout(r, 10));
            this.refill();
        }
        
        this.tokens--;
    }
    
    private refill(): void {
        const now = Date.now();
        const elapsed = (now - this.lastRefill) / 1000;
        this.tokens = Math.min(this.maxTokens, this.tokens + elapsed * this.refillRate);
        this.lastRefill = now;
    }
}
 
async function rateLimitedStream(
    call: ServerWritableStream<Data>,
    messagesPerSecond: number
): Promise<void> {
    const limiter = new RateLimitedProducer(messagesPerSecond, messagesPerSecond);
    
    for (const item of hugeDataset) {
        await limiter.acquire();  // Wait for rate limit token
        
        if (call.cancelled) return;
        
        const canContinue = call.write(item);
        if (!canContinue) {
            await new Promise<void>(r => call.once('drain', r));
        }
    }
    
    call.end();
}
 
// CONSUMER SIDE: Processing with controlled concurrency
async function controlledConsumer(
    stream: ClientReadableStream<WorkItem>,
    concurrency: number
): Promise<void> {
    const semaphore = new Semaphore(concurrency);
    const pending: Promise<void>[] = [];
    
    for await (const item of stream) {
        await semaphore.acquire();
        
        const task = processItem(item)
            .finally(() => semaphore.release());
        
        pending.push(task);
    }
    
    await Promise.all(pending);
}
 
class Semaphore {
    private permits: number;
    private readonly waiters: (() => void)[] = [];
    
    constructor(permits: number) {
        this.permits = permits;
    }
    
    async acquire(): Promise<void> {
        if (this.permits > 0) {
            this.permits--;
            return;
        }
        
        await new Promise<void>(resolve => {
            this.waiters.push(resolve);
        });
    }
    
    release(): void {
        const waiter = this.waiters.shift();
        if (waiter) {
            waiter();
        } else {
            this.permits++;
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137

// Comprehensive Stream Error Handling
 
import { status, Metadata, ClientReadableStream, ClientWritableStream } from '@grpc/grpc-js';
 
// Server-side: Graceful error handling
async function serverStreamWithErrorHandling(
    call: ServerWritableStream<Request, Response>
): Promise<void> {
    let itemsSent = 0;
    
    try {
        for (const item of generateItems()) {
            // Check for client cancellation before expensive work
            if (call.cancelled) {
                console.log(`Client cancelled after ${itemsSent} items`);
                return;  // No need to do anything, client is gone
            }
            
            try {
                const response = await processItem(item);
                
                const canContinue = call.write(response);
                if (!canContinue) {
                    await new Promise<void>(r => call.once('drain', r));
                }
                
                itemsSent++;
                
            } catch (itemError) {
                // Option 1: Skip failed item, continue stream
                console.error(`Skipping item: ${itemError.message}`);
                continue;
                
                // Option 2: End stream with error
                // call.destroy({
                //     code: status.INTERNAL,
                //     message: `Failed processing item: ${itemError.message}`,
                // });
                // return;
            }
        }
        
        // Successful completion
        call.end();
        
    } catch (fatalError) {
        // Unrecoverable error
        console.error(`Fatal stream error: ${fatalError.message}`);
        
        call.destroy({
            code: status.INTERNAL,
            message: 'Stream processing failed',
            details: fatalError.message,
        });
    }
}
 
// Client-side: Resilient stream consumption
async function resilientClientStream(
    request: Request,
    options: {
        maxRetries: number;
        retryDelay: number;
        onProgress?: (count: number) => void;
    }
): Promise<Response[]> {
    const { maxRetries, retryDelay, onProgress } = options;
    const results: Response[] = [];
    let retries = 0;
    let lastItemId: string | undefined;
    
    while (retries <= maxRetries) {
        try {
            const stream = client.listItems({
                ...request,
                resumeFromId: lastItemId,  // Resume from last successful item
            });
            
            for await (const item of stream) {
                results.push(item);
                lastItemId = item.id;
                onProgress?.(results.length);
            }
            
            // Stream completed successfully
            return results;
            
        } catch (error) {
            if (isRetryableError(error)) {
                retries++;
                console.log(`Stream failed (attempt ${retries}): ${error.message}`);
                
                if (retries <= maxRetries) {
                    await sleep(retryDelay * retries);  // Exponential backoff
                    continue;
                }
            }
            
            // Non-retryable or exhausted retries
            throw new Error(
                `Stream failed after ${retries} retries with ${results.length} items: ${error.message}`
            );
        }
    }
    
    return results;
}
 
function isRetryableError(error: any): boolean {
    const retryableCodes = [
        status.UNAVAILABLE,      // Transient failure
        status.RESOURCE_EXHAUSTED,  // Rate limited
        status.ABORTED,          // Conflict, retry
        status.INTERNAL,         // Server error (maybe)
    ];
    
    return retryableCodes.includes(error.code);
}
 
// Error classification for logging and metrics
function classifyStreamError(error: any): 'client' | 'server' | 'network' | 'timeout' {
    switch (error.code) {
        case status.CANCELLED:
            return 'client';
        case status.DEADLINE_EXCEEDED:
            return 'timeout';
        case status.UNAVAILABLE:
        case status.UNKNOWN:
            return 'network';
        case status.INTERNAL:
        case status.UNIMPLEMENTED:
        case status.RESOURCE_EXHAUSTED:
            return 'server';
        default:
            return 'client';  // Assume client error
    }
}