개념 및 통합 가이드

git clone https://github.com/chainstream-io/streaming_protobuf.git

git submodule add https://github.com/chainstream-io/streaming_protobuf.git

pip install kafka-python protobuf

from kafka import KafkaConsumer
from common import trade_event_pb2  # streaming_protobuf 리포지토리에서 가져오기

# 컨슈머 생성
consumer = KafkaConsumer(
    'eth.dex.trades',
    bootstrap_servers=['<your_broker_address>'],
    security_protocol='SASL_SSL',
    sasl_mechanism='SCRAM-SHA-512',
    sasl_plain_username='your_username',
    sasl_plain_password='your_password',
    auto_offset_reset='latest',
    enable_auto_commit=False,
    group_id='my-dex-consumer'
)

# 메시지 소비
for message in consumer:
    # protobuf 메시지 파싱
    trade_events = trade_event_pb2.TradeEvents()
    trade_events.ParseFromString(message.value)
    
    # DEX 거래 정보 출력
    for event in trade_events.events:
        print(f"Pool: {event.trade.pool_address}")
        print(f"Token A: {event.trade.token_a_address}")
        print(f"Token B: {event.trade.token_b_address}")
        print(f"Amount A: {event.trade.user_a_amount}")
        print(f"Amount B: {event.trade.user_b_amount}")
        print(f"Block: {event.block.height}")
        print("---")