cubestechdocsdashboard ↗

Enhanced Streams Quickstart

1. Download the proto

bash
curl -O https://cubestech.online/protos/enhanced-streams.proto

2. Generate Python code

bash
python -m pip install grpcio grpcio-tools protobuf

python -m grpc_tools.protoc \
  -I . \
  --python_out=. \
  --grpc_python_out=. \
  enhanced-streams.proto

Rename the generated module imports if your local generator replaces hyphens with underscores.

3. List topics

python
import grpc
import enhanced_streams_pb2 as pb
import enhanced_streams_pb2_grpc as rpc

channel = grpc.secure_channel(
    "streams.sol.cubestech.online:443:443",
    grpc.ssl_channel_credentials(),
)
client = rpc.StreamServiceStub(channel)
metadata = [("x-api-key", "YOUR_API_KEY")]

response = client.ListTopics(pb.ListTopicsRequest(), metadata=metadata)
for topic in response.topics:
    print(topic.name, topic.proto_message_type)

4. Subscribe

python
import json

request = pb.SubscribeRequest(
    topic="prod.rpc.solana.system.transfers",
    format=pb.JSON,
)

for message in client.Subscribe(request, metadata=metadata):
    payload = json.loads(message.payload)
    print({
        "partition": message.partition,
        "offset": message.offset,
        "timestamp_ms": message.timestamp_ms,
        "payload": payload,
    })

5. Store progress

Write your event and its partition-offset pair in one database transaction.

After a disconnect, open a new stream. Enhanced Streams do not expose a client offset in SubscribeRequest, so downstream deduplication must use event identity and stored metadata.