StreamData gRPC Method

Parameters

request

stream<SubscribeRequest>

Loading...

Returns

stream

stream<SubscribeUpdate>

Loading...

data

StreamResponse

Loading...

pong

Pong

Loading...

Request

1
package main
2

3
import (
4
	"context"
5
	"crypto/tls"
6
	"encoding/json"
7
	"fmt"
8
	"io"
9
	"log"
10
	"time"
11

12
	"google.golang.org/grpc"
13
	"google.golang.org/grpc/credentials"
14
	"google.golang.org/grpc/metadata"
15

16
	pb "hyperliquid-grpc-go/pb"
17
)
18

19
const (
20
	// Using Quicknode endpoint
21
	grpcEndpoint = "your-grpc-endpoint:port"
22
	authToken    = "your-auth-token"
23
)
24

25
// Create gRPC connection with TLS
26
func createConnection() (*grpc.ClientConn, error) {
27
	tlsConfig := &tls.Config{
28
		InsecureSkipVerify: false, // Set to true only for testing
29
	}
30
	creds := credentials.NewTLS(tlsConfig)
31

32
	conn, err := grpc.Dial(
33
		grpcEndpoint,
34
		grpc.WithTransportCredentials(creds),
35
		grpc.WithDefaultCallOptions(
36
			grpc.MaxCallRecvMsgSize(100*1024*1024), // 100MB
37
		),
38
	)
39

40
	return conn, err
41
}
42

43
// Create context with auth metadata
44
func createContext() context.Context {
45
	md := metadata.Pairs("x-token", authToken)
46
	return metadata.NewOutgoingContext(context.Background(), md)
47
}
48

49

50
// Stream trades
51
func streamTrades(client pb.StreamingClient) error {
52
	ctx := createContext()
53

54
	stream, err := client.StreamData(ctx)
55
	if err != nil {
56
		return fmt.Errorf("failed to create stream: %v", err)
57
	}
58

59
	// Send subscription request
60
	err = stream.Send(&pb.SubscribeRequest{
61
		Request: &pb.SubscribeRequest_Subscribe{
62
			Subscribe: &pb.StreamSubscribe{
63
				StreamType:    pb.StreamType_TRADES,
64
				StartBlock:    0,
65
				UserAddresses: []string{},
66
				Coins:         []string{},
67
			},
68
		},
69
	})
70
	if err != nil {
71
		return fmt.Errorf("failed to subscribe: %v", err)
72
	}
73

74
	// Start ping goroutine
75
	go func() {
76
		ticker := time.NewTicker(30 * time.Second)
77
		defer ticker.Stop()
78

79
		for range ticker.C {
80
			err := stream.Send(&pb.SubscribeRequest{
81
				Request: &pb.SubscribeRequest_Ping{
82
					Ping: &pb.Ping{
83
						Timestamp: time.Now().UnixMilli(),
84
					},
85
				},
86
			})
87
			if err != nil {
88
				log.Printf("Failed to send ping: %v", err)
89
				return
90
			}
91
		}
92
	}()
93

94
	// Receive messages
95
	log.Println("🔄 Starting trades stream...")
96
	for {
97
		msg, err := stream.Recv()
98
		if err == io.EOF {
99
			log.Println("Stream ended")
100
			break
101
		}
102
		if err != nil {
103
			return fmt.Errorf("receive error: %v", err)
104
		}
105

106
		switch update := msg.Update.(type) {
107
		case *pb.SubscribeUpdate_Data:
108
			var data map[string]interface{}
109
			if err := json.Unmarshal([]byte(update.Data.Data), &data); err != nil {
110
				log.Printf("⚠️  Failed to parse data: %v", err)
111
				continue
112
			}
113

114
			trades, ok := data["trades"].([]interface{})
115
			tradesCount := 0
116
			if ok {
117
				tradesCount = len(trades)
118
			}
119

120
			log.Printf("📈 Block %d: %d trades", update.Data.BlockNumber, tradesCount)
121
			
122
			// Pretty print first trade if available
123
			if tradesCount > 0 {
124
				firstTrade := trades[0]
125
				tradeJSON, _ := json.MarshalIndent(firstTrade, "   ", "  ")
126
				fmt.Printf("   Example trade:\n%s\n", tradeJSON)
127
			}
128

129
		case *pb.SubscribeUpdate_Pong:
130
			log.Printf("🏓 Pong received: %d", update.Pong.Timestamp)
131
		}
132
	}
133

134
	return nil
135
}
136

137
// Stream filtered trades
138
func streamFilteredTrades(client pb.StreamingClient, coins []string) error {
139
	ctx := createContext()
140

141
	stream, err := client.StreamData(ctx)
142
	if err != nil {
143
		return fmt.Errorf("failed to create stream: %v", err)
144
	}
145

146
	// Send filtered subscription
147
	err = stream.Send(&pb.SubscribeRequest{
148
		Request: &pb.SubscribeRequest_Subscribe{
149
			Subscribe: &pb.StreamSubscribe{
150
				StreamType:    pb.StreamType_TRADES,
151
				StartBlock:    0,
152
				UserAddresses: []string{},
153
				Coins:         coins,
154
			},
155
		},
156
	})
157
	if err != nil {
158
		return fmt.Errorf("failed to subscribe: %v", err)
159
	}
160

161
	log.Printf("🔄 Starting filtered trades stream for coins: %v", coins)
162

163
	// Receive messages
164
	for {
165
		msg, err := stream.Recv()
166
		if err == io.EOF {
167
			log.Println("Filtered stream ended")
168
			break
169
		}
170
		if err != nil {
171
			return fmt.Errorf("receive error: %v", err)
172
		}
173

174
		switch update := msg.Update.(type) {
175
		case *pb.SubscribeUpdate_Data:
176
			var data map[string]interface{}
177
			if err := json.Unmarshal([]byte(update.Data.Data), &data); err != nil {
178
				log.Printf("⚠️  Failed to parse data: %v", err)
179
				continue
180
			}
181

182
			if trades, ok := data["trades"].([]interface{}); ok && len(trades) > 0 {
183
				log.Printf("📈 Filtered Block %d: %d trades", update.Data.BlockNumber, len(trades))
184
				prettyJSON, _ := json.MarshalIndent(data, "", "  ")
185
				fmt.Println(string(prettyJSON))
186
			}
187
		}
188
	}
189

190
	return nil
191
}
192

193
// Stream orders
194
func streamOrders(client pb.StreamingClient) error {
195
	ctx := createContext()
196

197
	stream, err := client.StreamData(ctx)
198
	if err != nil {
199
		return fmt.Errorf("failed to create stream: %v", err)
200
	}
201

202
	// Send subscription request for orders
203
	err = stream.Send(&pb.SubscribeRequest{
204
		Request: &pb.SubscribeRequest_Subscribe{
205
			Subscribe: &pb.StreamSubscribe{
206
				StreamType:    pb.StreamType_ORDERS,
207
				StartBlock:    0,
208
				UserAddresses: []string{},
209
				Coins:         []string{},
210
			},
211
		},
212
	})
213
	if err != nil {
214
		return fmt.Errorf("failed to subscribe: %v", err)
215
	}
216

217
	log.Println("🔄 Starting orders stream...")
218

219
	// Receive messages
220
	for {
221
		msg, err := stream.Recv()
222
		if err == io.EOF {
223
			log.Println("Orders stream ended")
224
			break
225
		}
226
		if err != nil {
227
			return fmt.Errorf("receive error: %v", err)
228
		}
229

230
		switch update := msg.Update.(type) {
231
		case *pb.SubscribeUpdate_Data:
232
			var data map[string]interface{}
233
			if err := json.Unmarshal([]byte(update.Data.Data), &data); err != nil {
234
				log.Printf("⚠️  Failed to parse order data: %v", err)
235
				continue
236
			}
237

238
			log.Printf("📋 Orders Block %d:", update.Data.BlockNumber)
239
			prettyJSON, _ := json.MarshalIndent(data, "", "  ")
240
			fmt.Println(string(prettyJSON))
241
		}
242
	}
243

244
	return nil
245
}
246

247

248
// Interactive menu
249
func runInteractiveMenu(conn *grpc.ClientConn) {
250
	client := pb.NewStreamingClient(conn)
251

252
	fmt.Println("\n🎉 Connected successfully!")
253
	fmt.Println("\nAvailable streams:")
254
	fmt.Println("1. All trades")
255
	fmt.Println("2. Filtered trades (BTC, ETH)")
256
	fmt.Println("3. Orders")
257
	fmt.Println("4. Exit")
258

259
	for {
260
		fmt.Print("\nSelect stream (1-4): ")
261
		var choice string
262
		fmt.Scanln(&choice)
263

264
		switch choice {
265
		case "1":
266
			if err := streamTrades(client); err != nil {
267
				log.Printf("❌ Stream error: %v", err)
268
			}
269
		case "2":
270
			if err := streamFilteredTrades(client, []string{"BTC", "ETH"}); err != nil {
271
				log.Printf("❌ Filtered stream error: %v", err)
272
			}
273
		case "3":
274
			if err := streamOrders(client); err != nil {
275
				log.Printf("❌ Orders stream error: %v", err)
276
			}
277
		case "4":
278
			fmt.Println("👋 Goodbye!")
279
			return
280
		default:
281
			fmt.Println("❌ Invalid choice. Please select 1-4.")
282
		}
283
	}
284
}
285

286
func main() {
287
	// Create connection
288
	conn, err := createConnection()
289
	if err != nil {
290
		log.Fatalf("❌ Failed to connect: %v", err)
291
	}
292
	defer conn.Close()
293

294
	fmt.Printf("Connected to: %s\n", grpcEndpoint)
295

296
	// Run interactive menu
297
	runInteractiveMenu(conn)
298
}

1
package main
2

3
import (
4
	"context"
5
	"crypto/tls"
6
	"encoding/json"
7
	"fmt"
8
	"io"
9
	"log"
10
	"time"
11

12
	"google.golang.org/grpc"
13
	"google.golang.org/grpc/credentials"
14
	"google.golang.org/grpc/metadata"
15

16
	pb "hyperliquid-grpc-go/pb"
17
)
18

19
const (
20
	// Using Quicknode endpoint
21
	grpcEndpoint = "your-grpc-endpoint:port"
22
	authToken    = "your-auth-token"
23
)
24

25
// Create gRPC connection with TLS
26
func createConnection() (*grpc.ClientConn, error) {
27
	tlsConfig := &tls.Config{
28
		InsecureSkipVerify: false, // Set to true only for testing
29
	}
30
	creds := credentials.NewTLS(tlsConfig)
31

32
	conn, err := grpc.Dial(
33
		grpcEndpoint,
34
		grpc.WithTransportCredentials(creds),
35
		grpc.WithDefaultCallOptions(
36
			grpc.MaxCallRecvMsgSize(100*1024*1024), // 100MB
37
		),
38
	)
39

40
	return conn, err
41
}
42

43
// Create context with auth metadata
44
func createContext() context.Context {
45
	md := metadata.Pairs("x-token", authToken)
46
	return metadata.NewOutgoingContext(context.Background(), md)
47
}
48

49

50
// Stream trades
51
func streamTrades(client pb.StreamingClient) error {
52
	ctx := createContext()
53

54
	stream, err := client.StreamData(ctx)
55
	if err != nil {
56
		return fmt.Errorf("failed to create stream: %v", err)
57
	}
58

59
	// Send subscription request
60
	err = stream.Send(&pb.SubscribeRequest{
61
		Request: &pb.SubscribeRequest_Subscribe{
62
			Subscribe: &pb.StreamSubscribe{
63
				StreamType:    pb.StreamType_TRADES,
64
				StartBlock:    0,
65
				UserAddresses: []string{},
66
				Coins:         []string{},
67
			},
68
		},
69
	})
70
	if err != nil {
71
		return fmt.Errorf("failed to subscribe: %v", err)
72
	}
73

74
	// Start ping goroutine
75
	go func() {
76
		ticker := time.NewTicker(30 * time.Second)
77
		defer ticker.Stop()
78

79
		for range ticker.C {
80
			err := stream.Send(&pb.SubscribeRequest{
81
				Request: &pb.SubscribeRequest_Ping{
82
					Ping: &pb.Ping{
83
						Timestamp: time.Now().UnixMilli(),
84
					},
85
				},
86
			})
87
			if err != nil {
88
				log.Printf("Failed to send ping: %v", err)
89
				return
90
			}
91
		}
92
	}()
93

94
	// Receive messages
95
	log.Println("🔄 Starting trades stream...")
96
	for {
97
		msg, err := stream.Recv()
98
		if err == io.EOF {
99
			log.Println("Stream ended")
100
			break
101
		}
102
		if err != nil {
103
			return fmt.Errorf("receive error: %v", err)
104
		}
105

106
		switch update := msg.Update.(type) {
107
		case *pb.SubscribeUpdate_Data:
108
			var data map[string]interface{}
109
			if err := json.Unmarshal([]byte(update.Data.Data), &data); err != nil {
110
				log.Printf("⚠️  Failed to parse data: %v", err)
111
				continue
112
			}
113

114
			trades, ok := data["trades"].([]interface{})
115
			tradesCount := 0
116
			if ok {
117
				tradesCount = len(trades)
118
			}
119

120
			log.Printf("📈 Block %d: %d trades", update.Data.BlockNumber, tradesCount)
121
			
122
			// Pretty print first trade if available
123
			if tradesCount > 0 {
124
				firstTrade := trades[0]
125
				tradeJSON, _ := json.MarshalIndent(firstTrade, "   ", "  ")
126
				fmt.Printf("   Example trade:\n%s\n", tradeJSON)
127
			}
128

129
		case *pb.SubscribeUpdate_Pong:
130
			log.Printf("🏓 Pong received: %d", update.Pong.Timestamp)
131
		}
132
	}
133

134
	return nil
135
}
136

137
// Stream filtered trades
138
func streamFilteredTrades(client pb.StreamingClient, coins []string) error {
139
	ctx := createContext()
140

141
	stream, err := client.StreamData(ctx)
142
	if err != nil {
143
		return fmt.Errorf("failed to create stream: %v", err)
144
	}
145

146
	// Send filtered subscription
147
	err = stream.Send(&pb.SubscribeRequest{
148
		Request: &pb.SubscribeRequest_Subscribe{
149
			Subscribe: &pb.StreamSubscribe{
150
				StreamType:    pb.StreamType_TRADES,
151
				StartBlock:    0,
152
				UserAddresses: []string{},
153
				Coins:         coins,
154
			},
155
		},
156
	})
157
	if err != nil {
158
		return fmt.Errorf("failed to subscribe: %v", err)
159
	}
160

161
	log.Printf("🔄 Starting filtered trades stream for coins: %v", coins)
162

163
	// Receive messages
164
	for {
165
		msg, err := stream.Recv()
166
		if err == io.EOF {
167
			log.Println("Filtered stream ended")
168
			break
169
		}
170
		if err != nil {
171
			return fmt.Errorf("receive error: %v", err)
172
		}
173

174
		switch update := msg.Update.(type) {
175
		case *pb.SubscribeUpdate_Data:
176
			var data map[string]interface{}
177
			if err := json.Unmarshal([]byte(update.Data.Data), &data); err != nil {
178
				log.Printf("⚠️  Failed to parse data: %v", err)
179
				continue
180
			}
181

182
			if trades, ok := data["trades"].([]interface{}); ok && len(trades) > 0 {
183
				log.Printf("📈 Filtered Block %d: %d trades", update.Data.BlockNumber, len(trades))
184
				prettyJSON, _ := json.MarshalIndent(data, "", "  ")
185
				fmt.Println(string(prettyJSON))
186
			}
187
		}
188
	}
189

190
	return nil
191
}
192

193
// Stream orders
194
func streamOrders(client pb.StreamingClient) error {
195
	ctx := createContext()
196

197
	stream, err := client.StreamData(ctx)
198
	if err != nil {
199
		return fmt.Errorf("failed to create stream: %v", err)
200
	}
201

202
	// Send subscription request for orders
203
	err = stream.Send(&pb.SubscribeRequest{
204
		Request: &pb.SubscribeRequest_Subscribe{
205
			Subscribe: &pb.StreamSubscribe{
206
				StreamType:    pb.StreamType_ORDERS,
207
				StartBlock:    0,
208
				UserAddresses: []string{},
209
				Coins:         []string{},
210
			},
211
		},
212
	})
213
	if err != nil {
214
		return fmt.Errorf("failed to subscribe: %v", err)
215
	}
216

217
	log.Println("🔄 Starting orders stream...")
218

219
	// Receive messages
220
	for {
221
		msg, err := stream.Recv()
222
		if err == io.EOF {
223
			log.Println("Orders stream ended")
224
			break
225
		}
226
		if err != nil {
227
			return fmt.Errorf("receive error: %v", err)
228
		}
229

230
		switch update := msg.Update.(type) {
231
		case *pb.SubscribeUpdate_Data:
232
			var data map[string]interface{}
233
			if err := json.Unmarshal([]byte(update.Data.Data), &data); err != nil {
234
				log.Printf("⚠️  Failed to parse order data: %v", err)
235
				continue
236
			}
237

238
			log.Printf("📋 Orders Block %d:", update.Data.BlockNumber)
239
			prettyJSON, _ := json.MarshalIndent(data, "", "  ")
240
			fmt.Println(string(prettyJSON))
241
		}
242
	}
243

244
	return nil
245
}
246

247

248
// Interactive menu
249
func runInteractiveMenu(conn *grpc.ClientConn) {
250
	client := pb.NewStreamingClient(conn)
251

252
	fmt.Println("\n🎉 Connected successfully!")
253
	fmt.Println("\nAvailable streams:")
254
	fmt.Println("1. All trades")
255
	fmt.Println("2. Filtered trades (BTC, ETH)")
256
	fmt.Println("3. Orders")
257
	fmt.Println("4. Exit")
258

259
	for {
260
		fmt.Print("\nSelect stream (1-4): ")
261
		var choice string
262
		fmt.Scanln(&choice)
263

264
		switch choice {
265
		case "1":
266
			if err := streamTrades(client); err != nil {
267
				log.Printf("❌ Stream error: %v", err)
268
			}
269
		case "2":
270
			if err := streamFilteredTrades(client, []string{"BTC", "ETH"}); err != nil {
271
				log.Printf("❌ Filtered stream error: %v", err)
272
			}
273
		case "3":
274
			if err := streamOrders(client); err != nil {
275
				log.Printf("❌ Orders stream error: %v", err)
276
			}
277
		case "4":
278
			fmt.Println("👋 Goodbye!")
279
			return
280
		default:
281
			fmt.Println("❌ Invalid choice. Please select 1-4.")
282
		}
283
	}
284
}
285

286
func main() {
287
	// Create connection
288
	conn, err := createConnection()
289
	if err != nil {
290
		log.Fatalf("❌ Failed to connect: %v", err)
291
	}
292
	defer conn.Close()
293

294
	fmt.Printf("Connected to: %s\n", grpcEndpoint)
295

296
	// Run interactive menu
297
	runInteractiveMenu(conn)
298
}

1
const grpc = require('@grpc/grpc-js');
2
const protoLoader = require('@grpc/proto-loader');
3
const path = require('path');
4

5
// Configuration
6
const GRPC_ENDPOINT = 'your-grpc-endpoint:port';
7
const AUTH_TOKEN = 'your-auth-token';
8

9
// Load proto file
10
const packageDefinition = protoLoader.loadSync(
11
  path.join(__dirname, 'proto/streaming.proto'),
12
  {
13
    keepCase: true,
14
    longs: String,
15
    enums: String,
16
    defaults: true,
17
    oneofs: true,
18
  }
19
);
20

21
const hyperliquidProto = grpc.loadPackageDefinition(packageDefinition).hyperliquid;
22

23
// Create gRPC client with TLS
24
function createClient() {
25
  const credentials = grpc.credentials.createSsl();
26

27
  const client = new hyperliquidProto.Streaming(
28
    GRPC_ENDPOINT,
29
    credentials,
30
    {
31
      'grpc.max_receive_message_length': 100 * 1024 * 1024, // 100MB
32
    }
33
  );
34

35
  return client;
36
}
37

38
// Create metadata with auth token
39
function createMetadata() {
40
  const metadata = new grpc.Metadata();
41
  metadata.add('x-token', AUTH_TOKEN);
42
  return metadata;
43
}
44

45
// Test connectivity with ping
46
function testPing(client) {
47
  return new Promise((resolve, reject) => {
48
    const metadata = createMetadata();
49

50
    client.Ping({ count: 1 }, metadata, (error, response) => {
51
      if (error) {
52
        reject(new Error(`Ping failed: ${error.message}`));
53
        return;
54
      }
55

56
      console.log('✅ Ping successful:', response);
57
      resolve(response);
58
    });
59
  });
60
}
61

62
// Stream trades data
63
function streamTrades(client) {
64
  return new Promise((resolve, reject) => {
65
    const metadata = createMetadata();
66

67
    const stream = client.StreamData(metadata);
68

69
    // Handle incoming messages
70
    stream.on('data', (response) => {
71
      if (response.data) {
72
        try {
73
          const data = JSON.parse(response.data.data);
74
          const trades = data.trades || [];
75

76
          console.log(`📈 Block ${response.data.block_number}: ${trades.length} trades`);
77

78
          // Pretty print first trade if available
79
          if (trades.length > 0) {
80
            console.log('   Example trade:', JSON.stringify(trades[0], null, 2));
81
          }
82
        } catch (error) {
83
          console.log('⚠️  Failed to parse data:', error.message);
84
        }
85
      } else if (response.pong) {
86
        console.log('🏓 Pong received:', response.pong.timestamp);
87
      }
88
    });
89

90
    stream.on('error', (error) => {
91
      console.error('❌ Stream error:', error.message);
92
      reject(error);
93
    });
94

95
    stream.on('end', () => {
96
      console.log('Stream ended');
97
      resolve();
98
    });
99

100
    // Send subscription request
101
    stream.write({
102
      subscribe: {
103
        stream_type: 'TRADES',
104
        start_block: 0,
105
        user_addresses: [],
106
        coins: []
107
      }
108
    });
109

110
    // Start ping interval to keep connection alive
111
    const pingInterval = setInterval(() => {
112
      stream.write({
113
        ping: {
114
          timestamp: Date.now()
115
        }
116
      });
117
    }, 30000);
118

119
    // Clean up on exit
120
    const cleanup = () => {
121
      clearInterval(pingInterval);
122
      stream.end();
123
    };
124

125
    process.on('SIGINT', cleanup);
126
    process.on('SIGTERM', cleanup);
127

128
    console.log('🔄 Starting trades stream...');
129
  });
130
}
131

132
// Stream orders data
133
function streamOrders(client) {
134
  return new Promise((resolve, reject) => {
135
    const metadata = createMetadata();
136

137
    const stream = client.StreamData(metadata);
138

139
    stream.on('data', (response) => {
140
      if (response.data) {
141
        try {
142
          const data = JSON.parse(response.data.data);
143
          console.log(`📋 Orders Block ${response.data.block_number}:`);
144
          console.log(JSON.stringify(data, null, 2));
145
        } catch (error) {
146
          console.log('⚠️  Failed to parse order data:', error.message);
147
        }
148
      }
149
    });
150

151
    stream.on('error', (error) => {
152
      console.error('❌ Stream error:', error.message);
153
      reject(error);
154
    });
155

156
    stream.on('end', () => {
157
      console.log('Orders stream ended');
158
      resolve();
159
    });
160

161
    // Send subscription request for orders
162
    stream.write({
163
      subscribe: {
164
        stream_type: 'ORDERS',
165
        start_block: 0,
166
        user_addresses: [],
167
        coins: []
168
      }
169
    });
170

171
    console.log('🔄 Starting orders stream...');
172
  });
173
}
174

175
// Interactive menu
176
function showMenu() {
177
  console.log('\n🎉 Connected successfully!');
178
  console.log('\nAvailable streams:');
179
  console.log('1. All trades');
180
  console.log('2. Filtered trades (BTC, ETH)');
181
  console.log('3. Orders');
182
  console.log('4. Test ping');
183
  console.log('5. Exit');
184

185
  const readline = require('readline');
186
  const rl = readline.createInterface({
187
    input: process.stdin,
188
    output: process.stdout
189
  });
190

191
  return new Promise((resolve) => {
192
    rl.question('\nSelect stream (1-5): ', (choice) => {
193
      rl.close();
194
      resolve(choice);
195
    });
196
  });
197
}
198

199
// Stream filtered trades
200
function streamFilteredTrades(client, coins) {
201
  return new Promise((resolve, reject) => {
202
    const metadata = createMetadata();
203

204
    const stream = client.StreamData(metadata);
205

206
    stream.on('data', (response) => {
207
      if (response.data) {
208
        try {
209
          const data = JSON.parse(response.data.data);
210
          const trades = data.trades || [];
211

212
          if (trades.length > 0) {
213
            console.log(`📈 Filtered Block ${response.data.block_number}: ${trades.length} trades`);
214
            console.log(JSON.stringify(data, null, 2));
215
          }
216
        } catch (error) {
217
          console.log('⚠️  Failed to parse data:', error.message);
218
        }
219
      }
220
    });
221

222
    stream.on('error', (error) => {
223
      console.error('❌ Stream error:', error.message);
224
      reject(error);
225
    });
226

227
    stream.on('end', () => {
228
      console.log('Filtered stream ended');
229
      resolve();
230
    });
231

232
    // Send filtered subscription
233
    stream.write({
234
      subscribe: {
235
        stream_type: 'TRADES',
236
        start_block: 0,
237
        user_addresses: [],
238
        coins: coins
239
      }
240
    });
241

242
    console.log(`🔄 Starting filtered trades stream for coins: ${coins.join(', ')}`);
243
  });
244
}
245

246
// Main function
247
async function main() {
248
  try {
249
    console.log(`Testing connection to: ${GRPC_ENDPOINT}`);
250

251
    // Create client
252
    const client = createClient();
253

254
    // Test initial connectivity
255
    await testPing(client);
256

257
    // Show interactive menu
258
    while (true) {
259
      const choice = await showMenu();
260

261
      switch (choice) {
262
        case '1':
263
          await streamTrades(client);
264
          break;
265
        case '2':
266
          await streamFilteredTrades(client, ['BTC', 'ETH']);
267
          break;
268
        case '3':
269
          await streamOrders(client);
270
          break;
271
        case '4':
272
          await testPing(client);
273
          break;
274
        case '5':
275
          console.log('👋 Goodbye!');
276
          client.close();
277
          return;
278
        default:
279
          console.log('❌ Invalid choice. Please select 1-5.');
280
      }
281
    }
282

283
  } catch (error) {
284
    console.error('❌ Error:', error.message);
285
    process.exit(1);
286
  }
287
}
288

289
// Run the application
290
main();

1
const grpc = require('@grpc/grpc-js');
2
const protoLoader = require('@grpc/proto-loader');
3
const path = require('path');
4

5
// Configuration
6
const GRPC_ENDPOINT = 'your-grpc-endpoint:port';
7
const AUTH_TOKEN = 'your-auth-token';
8

9
// Load proto file
10
const packageDefinition = protoLoader.loadSync(
11
  path.join(__dirname, 'proto/streaming.proto'),
12
  {
13
    keepCase: true,
14
    longs: String,
15
    enums: String,
16
    defaults: true,
17
    oneofs: true,
18
  }
19
);
20

21
const hyperliquidProto = grpc.loadPackageDefinition(packageDefinition).hyperliquid;
22

23
// Create gRPC client with TLS
24
function createClient() {
25
  const credentials = grpc.credentials.createSsl();
26

27
  const client = new hyperliquidProto.Streaming(
28
    GRPC_ENDPOINT,
29
    credentials,
30
    {
31
      'grpc.max_receive_message_length': 100 * 1024 * 1024, // 100MB
32
    }
33
  );
34

35
  return client;
36
}
37

38
// Create metadata with auth token
39
function createMetadata() {
40
  const metadata = new grpc.Metadata();
41
  metadata.add('x-token', AUTH_TOKEN);
42
  return metadata;
43
}
44

45
// Test connectivity with ping
46
function testPing(client) {
47
  return new Promise((resolve, reject) => {
48
    const metadata = createMetadata();
49

50
    client.Ping({ count: 1 }, metadata, (error, response) => {
51
      if (error) {
52
        reject(new Error(`Ping failed: ${error.message}`));
53
        return;
54
      }
55

56
      console.log('✅ Ping successful:', response);
57
      resolve(response);
58
    });
59
  });
60
}
61

62
// Stream trades data
63
function streamTrades(client) {
64
  return new Promise((resolve, reject) => {
65
    const metadata = createMetadata();
66

67
    const stream = client.StreamData(metadata);
68

69
    // Handle incoming messages
70
    stream.on('data', (response) => {
71
      if (response.data) {
72
        try {
73
          const data = JSON.parse(response.data.data);
74
          const trades = data.trades || [];
75

76
          console.log(`📈 Block ${response.data.block_number}: ${trades.length} trades`);
77

78
          // Pretty print first trade if available
79
          if (trades.length > 0) {
80
            console.log('   Example trade:', JSON.stringify(trades[0], null, 2));
81
          }
82
        } catch (error) {
83
          console.log('⚠️  Failed to parse data:', error.message);
84
        }
85
      } else if (response.pong) {
86
        console.log('🏓 Pong received:', response.pong.timestamp);
87
      }
88
    });
89

90
    stream.on('error', (error) => {
91
      console.error('❌ Stream error:', error.message);
92
      reject(error);
93
    });
94

95
    stream.on('end', () => {
96
      console.log('Stream ended');
97
      resolve();
98
    });
99

100
    // Send subscription request
101
    stream.write({
102
      subscribe: {
103
        stream_type: 'TRADES',
104
        start_block: 0,
105
        user_addresses: [],
106
        coins: []
107
      }
108
    });
109

110
    // Start ping interval to keep connection alive
111
    const pingInterval = setInterval(() => {
112
      stream.write({
113
        ping: {
114
          timestamp: Date.now()
115
        }
116
      });
117
    }, 30000);
118

119
    // Clean up on exit
120
    const cleanup = () => {
121
      clearInterval(pingInterval);
122
      stream.end();
123
    };
124

125
    process.on('SIGINT', cleanup);
126
    process.on('SIGTERM', cleanup);
127

128
    console.log('🔄 Starting trades stream...');
129
  });
130
}
131

132
// Stream orders data
133
function streamOrders(client) {
134
  return new Promise((resolve, reject) => {
135
    const metadata = createMetadata();
136

137
    const stream = client.StreamData(metadata);
138

139
    stream.on('data', (response) => {
140
      if (response.data) {
141
        try {
142
          const data = JSON.parse(response.data.data);
143
          console.log(`📋 Orders Block ${response.data.block_number}:`);
144
          console.log(JSON.stringify(data, null, 2));
145
        } catch (error) {
146
          console.log('⚠️  Failed to parse order data:', error.message);
147
        }
148
      }
149
    });
150

151
    stream.on('error', (error) => {
152
      console.error('❌ Stream error:', error.message);
153
      reject(error);
154
    });
155

156
    stream.on('end', () => {
157
      console.log('Orders stream ended');
158
      resolve();
159
    });
160

161
    // Send subscription request for orders
162
    stream.write({
163
      subscribe: {
164
        stream_type: 'ORDERS',
165
        start_block: 0,
166
        user_addresses: [],
167
        coins: []
168
      }
169
    });
170

171
    console.log('🔄 Starting orders stream...');
172
  });
173
}
174

175
// Interactive menu
176
function showMenu() {
177
  console.log('\n🎉 Connected successfully!');
178
  console.log('\nAvailable streams:');
179
  console.log('1. All trades');
180
  console.log('2. Filtered trades (BTC, ETH)');
181
  console.log('3. Orders');
182
  console.log('4. Test ping');
183
  console.log('5. Exit');
184

185
  const readline = require('readline');
186
  const rl = readline.createInterface({
187
    input: process.stdin,
188
    output: process.stdout
189
  });
190

191
  return new Promise((resolve) => {
192
    rl.question('\nSelect stream (1-5): ', (choice) => {
193
      rl.close();
194
      resolve(choice);
195
    });
196
  });
197
}
198

199
// Stream filtered trades
200
function streamFilteredTrades(client, coins) {
201
  return new Promise((resolve, reject) => {
202
    const metadata = createMetadata();
203

204
    const stream = client.StreamData(metadata);
205

206
    stream.on('data', (response) => {
207
      if (response.data) {
208
        try {
209
          const data = JSON.parse(response.data.data);
210
          const trades = data.trades || [];
211

212
          if (trades.length > 0) {
213
            console.log(`📈 Filtered Block ${response.data.block_number}: ${trades.length} trades`);
214
            console.log(JSON.stringify(data, null, 2));
215
          }
216
        } catch (error) {
217
          console.log('⚠️  Failed to parse data:', error.message);
218
        }
219
      }
220
    });
221

222
    stream.on('error', (error) => {
223
      console.error('❌ Stream error:', error.message);
224
      reject(error);
225
    });
226

227
    stream.on('end', () => {
228
      console.log('Filtered stream ended');
229
      resolve();
230
    });
231

232
    // Send filtered subscription
233
    stream.write({
234
      subscribe: {
235
        stream_type: 'TRADES',
236
        start_block: 0,
237
        user_addresses: [],
238
        coins: coins
239
      }
240
    });
241

242
    console.log(`🔄 Starting filtered trades stream for coins: ${coins.join(', ')}`);
243
  });
244
}
245

246
// Main function
247
async function main() {
248
  try {
249
    console.log(`Testing connection to: ${GRPC_ENDPOINT}`);
250

251
    // Create client
252
    const client = createClient();
253

254
    // Test initial connectivity
255
    await testPing(client);
256

257
    // Show interactive menu
258
    while (true) {
259
      const choice = await showMenu();
260

261
      switch (choice) {
262
        case '1':
263
          await streamTrades(client);
264
          break;
265
        case '2':
266
          await streamFilteredTrades(client, ['BTC', 'ETH']);
267
          break;
268
        case '3':
269
          await streamOrders(client);
270
          break;
271
        case '4':
272
          await testPing(client);
273
          break;
274
        case '5':
275
          console.log('👋 Goodbye!');
276
          client.close();
277
          return;
278
        default:
279
          console.log('❌ Invalid choice. Please select 1-5.');
280
      }
281
    }
282

283
  } catch (error) {
284
    console.error('❌ Error:', error.message);
285
    process.exit(1);
286
  }
287
}
288

289
// Run the application
290
main();

1
import grpc
2
import json
3
import time
4
import signal
5
import sys
6
import threading
7
from pb import streaming_pb2, streaming_pb2_grpc
8

9
# Configuration
10
GRPC_ENDPOINT = 'your-grpc-endpoint:port'
11
AUTH_TOKEN = 'your-auth-token'
12

13
def create_channel():
14
    """Create gRPC channel with TLS"""
15
    credentials = grpc.ssl_channel_credentials()
16
    channel = grpc.secure_channel(
17
        GRPC_ENDPOINT,
18
        credentials,
19
        options=[
20
            ('grpc.max_receive_message_length', 100 * 1024 * 1024),  # 100MB
21
            ('grpc.max_send_message_length', 100 * 1024 * 1024),     # 100MB
22
        ]
23
    )
24
    return channel
25

26
def create_metadata():
27
    """Create metadata with auth token"""
28
    return [('x-token', AUTH_TOKEN)]
29

30
def test_ping(stub):
31
    """Test connectivity with ping"""
32
    try:
33
        metadata = create_metadata()
34
        request = streaming_pb2.PingRequest(count=1)
35

36
        response = stub.Ping(request, metadata=metadata)
37
        print(f"✅ Ping successful: {response}")
38
        return True
39
    except grpc.RpcError as e:
40
        print(f"❌ Ping failed: {e.details()}")
41
        return False
42

43
def stream_trades(stub):
44
    """Stream trades data"""
45
    try:
46
        metadata = create_metadata()
47

48
        print("🔄 Starting trades stream...")
49

50
        # Create request generator for bidirectional stream
51
        def request_generator():
52
            # Send initial subscription request
53
            subscribe_request = streaming_pb2.SubscribeRequest()
54
            subscribe_request.subscribe.stream_type = streaming_pb2.StreamType.TRADES
55
            subscribe_request.subscribe.start_block = 0
56
            # Empty lists for user_addresses and coins means get all
57
            yield subscribe_request
58

59
            # Keep sending pings to maintain connection
60
            while True:
61
                time.sleep(30)
62
                ping_request = streaming_pb2.SubscribeRequest()
63
                ping_request.ping.timestamp = int(time.time() * 1000)
64
                yield ping_request
65

66
        # Create bidirectional stream
67
        stream = stub.StreamData(request_generator(), metadata=metadata)
68

69
        # Handle stream responses
70
        for response in stream:
71
            try:
72
                if response.HasField('data'):
73
                    data = json.loads(response.data.data)
74
                    trades = data.get('trades', [])
75

76
                    print(f"📈 Block {response.data.block_number}: {len(trades)} trades")
77

78
                    # Pretty print first trade if available
79
                    if trades:
80
                        print(f"   Example trade: {json.dumps(trades[0], indent=2)}")
81

82
                elif response.HasField('pong'):
83
                    print(f"🏓 Pong received: {response.pong.timestamp}")
84

85
            except json.JSONDecodeError as e:
86
                print(f"⚠️  Failed to parse data: {e}")
87
            except KeyboardInterrupt:
88
                print("\n⏹️  Stream interrupted by user")
89
                break
90

91
    except grpc.RpcError as e:
92
        print(f"❌ Stream error: {e.details()}")
93
    except KeyboardInterrupt:
94
        print("\n⏹️  Stream interrupted by user")
95

96
def stream_orders(stub):
97
    """Stream orders data"""
98
    try:
99
        metadata = create_metadata()
100

101
        print("🔄 Starting orders stream...")
102

103
        # Create request generator for bidirectional stream
104
        def request_generator():
105
            # Send initial subscription request for orders
106
            subscribe_request = streaming_pb2.SubscribeRequest()
107
            subscribe_request.subscribe.stream_type = streaming_pb2.StreamType.ORDERS
108
            subscribe_request.subscribe.start_block = 0
109
            yield subscribe_request
110

111
            # Keep sending pings to maintain connection
112
            while True:
113
                time.sleep(30)
114
                ping_request = streaming_pb2.SubscribeRequest()
115
                ping_request.ping.timestamp = int(time.time() * 1000)
116
                yield ping_request
117

118
        # Create bidirectional stream
119
        stream = stub.StreamData(request_generator(), metadata=metadata)
120

121
        # Handle stream responses
122
        for response in stream:
123
            try:
124
                if response.HasField('data'):
125
                    data = json.loads(response.data.data)
126
                    print(f"📋 Orders Block {response.data.block_number}:")
127
                    print(json.dumps(data, indent=2))
128

129
            except json.JSONDecodeError as e:
130
                print(f"⚠️  Failed to parse order data: {e}")
131
            except KeyboardInterrupt:
132
                print("\n⏹️  Stream interrupted by user")
133
                break
134

135
    except grpc.RpcError as e:
136
        print(f"❌ Stream error: {e.details()}")
137
    except KeyboardInterrupt:
138
        print("\n⏹️  Stream interrupted by user")
139

140
def stream_filtered_trades(stub, coins):
141
    """Stream filtered trades for specific coins"""
142
    try:
143
        metadata = create_metadata()
144

145
        print(f"🔄 Starting filtered trades stream for coins: {', '.join(coins)}")
146

147
        # Create request generator for bidirectional stream
148
        def request_generator():
149
            # Send filtered subscription request
150
            subscribe_request = streaming_pb2.SubscribeRequest()
151
            subscribe_request.subscribe.stream_type = streaming_pb2.StreamType.TRADES
152
            subscribe_request.subscribe.start_block = 0
153
            subscribe_request.subscribe.coins.extend(coins)
154
            yield subscribe_request
155

156
            # Keep sending pings to maintain connection
157
            while True:
158
                time.sleep(30)
159
                ping_request = streaming_pb2.SubscribeRequest()
160
                ping_request.ping.timestamp = int(time.time() * 1000)
161
                yield ping_request
162

163
        # Create bidirectional stream
164
        stream = stub.StreamData(request_generator(), metadata=metadata)
165

166
        # Handle stream responses
167
        for response in stream:
168
            try:
169
                if response.HasField('data'):
170
                    data = json.loads(response.data.data)
171
                    trades = data.get('trades', [])
172

173
                    if trades:
174
                        print(f"📈 Filtered Block {response.data.block_number}: {len(trades)} trades")
175
                        print(json.dumps(data, indent=2))
176

177
            except json.JSONDecodeError as e:
178
                print(f"⚠️  Failed to parse data: {e}")
179
            except KeyboardInterrupt:
180
                print("\n⏹️  Stream interrupted by user")
181
                break
182

183
    except grpc.RpcError as e:
184
        print(f"❌ Stream error: {e.details()}")
185
    except KeyboardInterrupt:
186
        print("\n⏹️  Stream interrupted by user")
187

188

189
def show_menu():
190
    """Display interactive menu"""
191
    print("\n🎉 Connected successfully!")
192
    print("\nAvailable streams:")
193
    print("1. All trades")
194
    print("2. Filtered trades (BTC, ETH)")
195
    print("3. Orders")
196
    print("4. Test ping")
197
    print("5. Exit")
198

199
    while True:
200
        try:
201
            choice = input("\nSelect stream (1-5): ").strip()
202
            if choice in ['1', '2', '3', '4', '5']:
203
                return choice
204
            else:
205
                print("❌ Invalid choice. Please select 1-5.")
206
        except KeyboardInterrupt:
207
            return '5'
208

209
def main():
210
    """Main function"""
211
    try:
212
        print(f"Testing connection to: {GRPC_ENDPOINT}")
213

214
        # Create channel and stub
215
        channel = create_channel()
216
        streaming_stub = streaming_pb2_grpc.StreamingStub(channel)
217

218
        # Test initial connectivity
219
        if not test_ping(streaming_stub):
220
            print("❌ Connection test failed!")
221
            return
222

223
        # Interactive menu loop
224
        while True:
225
            choice = show_menu()
226

227
            if choice == '1':
228
                stream_trades(streaming_stub)
229
            elif choice == '2':
230
                stream_filtered_trades(streaming_stub, ['BTC', 'ETH'])
231
            elif choice == '3':
232
                stream_orders(streaming_stub)
233
            elif choice == '4':
234
                test_ping(streaming_stub)
235
            elif choice == '5':
236
                print("👋 Goodbye!")
237
                break
238

239
        # Close channel
240
        channel.close()
241

242
    except KeyboardInterrupt:
243
        print("\n👋 Goodbye!")
244
    except Exception as e:
245
        print(f"❌ Error: {e}")
246
        sys.exit(1)
247

248
if __name__ == "__main__":
249
    main()

1
import grpc
2
import json
3
import time
4
import signal
5
import sys
6
import threading
7
from pb import streaming_pb2, streaming_pb2_grpc
8

9
# Configuration
10
GRPC_ENDPOINT = 'your-grpc-endpoint:port'
11
AUTH_TOKEN = 'your-auth-token'
12

13
def create_channel():
14
    """Create gRPC channel with TLS"""
15
    credentials = grpc.ssl_channel_credentials()
16
    channel = grpc.secure_channel(
17
        GRPC_ENDPOINT,
18
        credentials,
19
        options=[
20
            ('grpc.max_receive_message_length', 100 * 1024 * 1024),  # 100MB
21
            ('grpc.max_send_message_length', 100 * 1024 * 1024),     # 100MB
22
        ]
23
    )
24
    return channel
25

26
def create_metadata():
27
    """Create metadata with auth token"""
28
    return [('x-token', AUTH_TOKEN)]
29

30
def test_ping(stub):
31
    """Test connectivity with ping"""
32
    try:
33
        metadata = create_metadata()
34
        request = streaming_pb2.PingRequest(count=1)
35

36
        response = stub.Ping(request, metadata=metadata)
37
        print(f"✅ Ping successful: {response}")
38
        return True
39
    except grpc.RpcError as e:
40
        print(f"❌ Ping failed: {e.details()}")
41
        return False
42

43
def stream_trades(stub):
44
    """Stream trades data"""
45
    try:
46
        metadata = create_metadata()
47

48
        print("🔄 Starting trades stream...")
49

50
        # Create request generator for bidirectional stream
51
        def request_generator():
52
            # Send initial subscription request
53
            subscribe_request = streaming_pb2.SubscribeRequest()
54
            subscribe_request.subscribe.stream_type = streaming_pb2.StreamType.TRADES
55
            subscribe_request.subscribe.start_block = 0
56
            # Empty lists for user_addresses and coins means get all
57
            yield subscribe_request
58

59
            # Keep sending pings to maintain connection
60
            while True:
61
                time.sleep(30)
62
                ping_request = streaming_pb2.SubscribeRequest()
63
                ping_request.ping.timestamp = int(time.time() * 1000)
64
                yield ping_request
65

66
        # Create bidirectional stream
67
        stream = stub.StreamData(request_generator(), metadata=metadata)
68

69
        # Handle stream responses
70
        for response in stream:
71
            try:
72
                if response.HasField('data'):
73
                    data = json.loads(response.data.data)
74
                    trades = data.get('trades', [])
75

76
                    print(f"📈 Block {response.data.block_number}: {len(trades)} trades")
77

78
                    # Pretty print first trade if available
79
                    if trades:
80
                        print(f"   Example trade: {json.dumps(trades[0], indent=2)}")
81

82
                elif response.HasField('pong'):
83
                    print(f"🏓 Pong received: {response.pong.timestamp}")
84

85
            except json.JSONDecodeError as e:
86
                print(f"⚠️  Failed to parse data: {e}")
87
            except KeyboardInterrupt:
88
                print("\n⏹️  Stream interrupted by user")
89
                break
90

91
    except grpc.RpcError as e:
92
        print(f"❌ Stream error: {e.details()}")
93
    except KeyboardInterrupt:
94
        print("\n⏹️  Stream interrupted by user")
95

96
def stream_orders(stub):
97
    """Stream orders data"""
98
    try:
99
        metadata = create_metadata()
100

101
        print("🔄 Starting orders stream...")
102

103
        # Create request generator for bidirectional stream
104
        def request_generator():
105
            # Send initial subscription request for orders
106
            subscribe_request = streaming_pb2.SubscribeRequest()
107
            subscribe_request.subscribe.stream_type = streaming_pb2.StreamType.ORDERS
108
            subscribe_request.subscribe.start_block = 0
109
            yield subscribe_request
110

111
            # Keep sending pings to maintain connection
112
            while True:
113
                time.sleep(30)
114
                ping_request = streaming_pb2.SubscribeRequest()
115
                ping_request.ping.timestamp = int(time.time() * 1000)
116
                yield ping_request
117

118
        # Create bidirectional stream
119
        stream = stub.StreamData(request_generator(), metadata=metadata)
120

121
        # Handle stream responses
122
        for response in stream:
123
            try:
124
                if response.HasField('data'):
125
                    data = json.loads(response.data.data)
126
                    print(f"📋 Orders Block {response.data.block_number}:")
127
                    print(json.dumps(data, indent=2))
128

129
            except json.JSONDecodeError as e:
130
                print(f"⚠️  Failed to parse order data: {e}")
131
            except KeyboardInterrupt:
132
                print("\n⏹️  Stream interrupted by user")
133
                break
134

135
    except grpc.RpcError as e:
136
        print(f"❌ Stream error: {e.details()}")
137
    except KeyboardInterrupt:
138
        print("\n⏹️  Stream interrupted by user")
139

140
def stream_filtered_trades(stub, coins):
141
    """Stream filtered trades for specific coins"""
142
    try:
143
        metadata = create_metadata()
144

145
        print(f"🔄 Starting filtered trades stream for coins: {', '.join(coins)}")
146

147
        # Create request generator for bidirectional stream
148
        def request_generator():
149
            # Send filtered subscription request
150
            subscribe_request = streaming_pb2.SubscribeRequest()
151
            subscribe_request.subscribe.stream_type = streaming_pb2.StreamType.TRADES
152
            subscribe_request.subscribe.start_block = 0
153
            subscribe_request.subscribe.coins.extend(coins)
154
            yield subscribe_request
155

156
            # Keep sending pings to maintain connection
157
            while True:
158
                time.sleep(30)
159
                ping_request = streaming_pb2.SubscribeRequest()
160
                ping_request.ping.timestamp = int(time.time() * 1000)
161
                yield ping_request
162

163
        # Create bidirectional stream
164
        stream = stub.StreamData(request_generator(), metadata=metadata)
165

166
        # Handle stream responses
167
        for response in stream:
168
            try:
169
                if response.HasField('data'):
170
                    data = json.loads(response.data.data)
171
                    trades = data.get('trades', [])
172

173
                    if trades:
174
                        print(f"📈 Filtered Block {response.data.block_number}: {len(trades)} trades")
175
                        print(json.dumps(data, indent=2))
176

177
            except json.JSONDecodeError as e:
178
                print(f"⚠️  Failed to parse data: {e}")
179
            except KeyboardInterrupt:
180
                print("\n⏹️  Stream interrupted by user")
181
                break
182

183
    except grpc.RpcError as e:
184
        print(f"❌ Stream error: {e.details()}")
185
    except KeyboardInterrupt:
186
        print("\n⏹️  Stream interrupted by user")
187

188

189
def show_menu():
190
    """Display interactive menu"""
191
    print("\n🎉 Connected successfully!")
192
    print("\nAvailable streams:")
193
    print("1. All trades")
194
    print("2. Filtered trades (BTC, ETH)")
195
    print("3. Orders")
196
    print("4. Test ping")
197
    print("5. Exit")
198

199
    while True:
200
        try:
201
            choice = input("\nSelect stream (1-5): ").strip()
202
            if choice in ['1', '2', '3', '4', '5']:
203
                return choice
204
            else:
205
                print("❌ Invalid choice. Please select 1-5.")
206
        except KeyboardInterrupt:
207
            return '5'
208

209
def main():
210
    """Main function"""
211
    try:
212
        print(f"Testing connection to: {GRPC_ENDPOINT}")
213

214
        # Create channel and stub
215
        channel = create_channel()
216
        streaming_stub = streaming_pb2_grpc.StreamingStub(channel)
217

218
        # Test initial connectivity
219
        if not test_ping(streaming_stub):
220
            print("❌ Connection test failed!")
221
            return
222

223
        # Interactive menu loop
224
        while True:
225
            choice = show_menu()
226

227
            if choice == '1':
228
                stream_trades(streaming_stub)
229
            elif choice == '2':
230
                stream_filtered_trades(streaming_stub, ['BTC', 'ETH'])
231
            elif choice == '3':
232
                stream_orders(streaming_stub)
233
            elif choice == '4':
234
                test_ping(streaming_stub)
235
            elif choice == '5':
236
                print("👋 Goodbye!")
237
                break
238

239
        # Close channel
240
        channel.close()
241

242
    except KeyboardInterrupt:
243
        print("\n👋 Goodbye!")
244
    except Exception as e:
245
        print(f"❌ Error: {e}")
246
        sys.exit(1)
247

248
if __name__ == "__main__":
249
    main()

Don't have an account yet?

Create your Quicknode endpoint in seconds and start building

Get started for free