Events & Subscriptions

package main

import (
    "context"
    "fmt"
    "io"
    "log"

    pb "path/to/proto"
    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials/insecure"
)

func main() {
    conn, err := grpc.Dial("localhost:5003", grpc.WithTransportCredentials(insecure.NewCredentials()))
    if err != nil {
        log.Fatalf("Failed to connect: %v", err)
    }
    defer conn.Close()

    client := pb.NewOrderbookServiceClient(conn)

    request := &pb.SubscribeMarketEventsRequest{
        Base: &pb.OrderbookCurrency{
            Network: &pb.Network{
                Protocol: pb.Protocol_PROTOCOL_BITCOIN,
                Id:       "0a03cf40",
            },
            AssetId: "0x0000000000000000000000000000000000000000000000000000000000000000",
        },
        Quote: &pb.OrderbookCurrency{
            Network: &pb.Network{
                Protocol: pb.Protocol_PROTOCOL_EVM,
                Id:       "11155111",
            },
            AssetId: "0x0000000000000000000000000000000000000000",
        },
    }

    stream, err := client.SubscribeMarketEvents(context.Background(), request)
    if err != nil {
        log.Fatalf("Failed to subscribe: %v", err)
    }

    for {
        event, err := stream.Recv()
        if err == io.EOF {
            fmt.Println("Stream ended")
            break
        }
        if err != nil {
            log.Fatalf("Stream error: %v", err)
        }

        if event.GetIsSynced() {
            fmt.Println("✓ Market data synced")
        }

        if update := event.GetOrderbookUpdate(); update != nil {
            updatedOrders := update.GetUpdatedOrders()
            removedOrders := update.GetRemovedOrders()

            fmt.Println("Orderbook updated:")
            fmt.Printf("  Modified: %d\n", len(updatedOrders))
            fmt.Printf("  Removed: %d\n", len(removedOrders))
        }

        if trade := event.GetTradeUpdate(); trade != nil {
            fmt.Println("Trade executed:")
            fmt.Printf("  Amount: %s\n", trade.GetBaseAmount())
            fmt.Printf("  Price: %s\n", trade.GetPrice())
            side := "SELL"
            if trade.GetMakerOrderSide() == pb.OrderSide_BUY {
                side = "BUY"
            }
            fmt.Printf("  Side: %s\n", side)
        }

        if stats := event.GetDailyStatsUpdate(); stats != nil {
            vol := stats.GetVolatility()
            fmt.Println("24h Stats:")
            fmt.Printf("  High: %s\n", vol.GetHighPrice())
            fmt.Printf("  Low: %s\n", vol.GetLowPrice())
            fmt.Printf("  Volume: %s\n", vol.GetBaseVolume())
        }

        if candleUpdate := event.GetCandlestickUpdate(); candleUpdate != nil {
            candle := candleUpdate.GetCandlestick()
            fmt.Printf("New candle: %v\n", candleUpdate.GetInterval())
            fmt.Printf("  O: %s\n", candle.GetOpen())
            fmt.Printf("  H: %s\n", candle.GetHigh())
            fmt.Printf("  L: %s\n", candle.GetLow())
            fmt.Printf("  C: %s\n", candle.GetClose())
        }
    }
}

use futures::stream::StreamExt;
use proto::orderbook_service_client::OrderbookServiceClient;
use proto::{Network, OrderbookCurrency, Protocol, SubscribeMarketEventsRequest};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut client = OrderbookServiceClient::connect("http://localhost:5003").await?;

    let request = SubscribeMarketEventsRequest {
        base: Some(OrderbookCurrency {
            network: Some(Network {
                protocol: Protocol::Bitcoin as i32,
                id: "0a03cf40".to_string(),
            }),
            asset_id: "0x0000000000000000000000000000000000000000000000000000000000000000".to_string(),
        }),
        quote: Some(OrderbookCurrency {
            network: Some(Network {
                protocol: Protocol::Evm as i32,
                id: "11155111".to_string(),
            }),
            asset_id: "0x0000000000000000000000000000000000000000".to_string(),
        }),
    };

    let mut stream = client.subscribe_market_events(request).await?.into_inner();

    while let Some(event) = stream.next().await {
        match event {
            Ok(event) => {
                if event.is_synced {
                    println!("✓ Market data synced");
                }

                if let Some(update) = &event.orderbook_update {
                    println!("Orderbook updated:");
                    println!("  Modified: {}", update.updated_orders.len());
                    println!("  Removed: {}", update.removed_orders.len());
                }

                if let Some(trade) = &event.trade_update {
                    println!("Trade executed:");
                    println!("  Amount: {}", trade.base_amount);
                    println!("  Price: {}", trade.price);
                    let side = if trade.maker_order_side == 0 { "BUY" } else { "SELL" };
                    println!("  Side: {}", side);
                }

                if let Some(stats) = &event.daily_stats_update {
                    if let Some(vol) = &stats.volatility {
                        println!("24h Stats:");
                        println!("  High: {}", vol.high_price);
                        println!("  Low: {}", vol.low_price);
                        println!("  Volume: {}", vol.base_volume);
                    }
                }

                if let Some(candle_update) = &event.candlestick_update {
                    if let Some(candle) = &candle_update.candlestick {
                        println!("New candle: {:?}", candle_update.interval);
                        println!("  O: {}", candle.open);
                        println!("  H: {}", candle.high);
                        println!("  L: {}", candle.low);
                        println!("  C: {}", candle.close);
                    }
                }
            }
            Err(e) => {
                eprintln!("Stream error: {}", e);
                break;
            }
        }
    }

    println!("Stream ended");
    Ok(())
}

package main

import (
    "context"
    "fmt"
    "io"
    "log"

    pb "path/to/proto"
    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials/insecure"
)

func main() {
    conn, err := grpc.Dial("localhost:5003", grpc.WithTransportCredentials(insecure.NewCredentials()))
    if err != nil {
        log.Fatalf("Failed to connect: %v", err)
    }
    defer conn.Close()

    client := pb.NewOrderbookServiceClient(conn)

    request := &pb.SubscribeDexEventsRequest{}
    stream, err := client.SubscribeDexEvents(context.Background(), request)
    if err != nil {
        log.Fatalf("Failed to subscribe: %v", err)
    }

    for {
        event, err := stream.Recv()
        if err == io.EOF {
            fmt.Println("Stream ended")
            break
        }
        if err != nil {
            log.Fatalf("Stream error: %v", err)
        }

        timestamp := event.GetTimestamp()

        if balance := event.GetBalanceUpdate(); balance != nil {
            fmt.Printf("Balance updated:\n")
            fmt.Printf("  Currency: %s\n", balance.GetCurrency().GetAssetId())
            fmt.Printf("  Available: %s\n", balance.GetBalance().GetSending())
        }

        if update := event.GetOrderUpdate(); update != nil {
            if created := update.GetOrderCreated(); created != nil {
                fmt.Printf("Order created: %s\n", created.GetOrderId())
            }

            if completed := update.GetOrderCompleted(); completed != nil {
                fmt.Printf("Order completed: %s\n", completed.GetOrderId())
            }

            if canceled := update.GetOrderCanceled(); canceled != nil {
                fmt.Printf("Order cancelled: %s\n", canceled.GetOrderId())
            }
        }

        if matched := event.GetOrderMatched(); matched != nil {
            fmt.Printf("Order matched:\n")
            fmt.Printf("  Order ID: %s\n", matched.GetOwnOrderId())
            fmt.Printf("  Is Taker: %v\n", matched.GetIsTaker())
            fmt.Printf("  Routes: %d\n", len(matched.GetSwapRoute()))
        }

        if swap := event.GetSwapUpdate(); swap != nil {
            progress := swap.GetProgress()
            fmt.Printf("Swap update:\n")
            fmt.Printf("  Swap ID: %s\n", swap.GetSwapId())
            fmt.Printf("  Status: %v\n", progress.GetStatus())
            fmt.Printf("  Receiving: %s\n", progress.GetReceivingAmount())
            fmt.Printf("  Paying: %s\n", progress.GetPayingAmount())

            if progress.GetError() != "" {
                fmt.Printf("  Error: %s\n", progress.GetError())
            }
        }

        if trade := event.GetMarketTradeUpdate(); trade != nil {
            fmt.Printf("Trade executed: %v\n", trade.GetTrade())
        }

        _ = timestamp // Use timestamp if needed
    }
}

use futures::stream::StreamExt;
use proto::orderbook_service_client::OrderbookServiceClient;
use proto::SubscribeDexEventsRequest;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut client = OrderbookServiceClient::connect("http://localhost:5003").await?;

    let request = SubscribeDexEventsRequest {};
    let mut stream = client.subscribe_dex_events(request).await?.into_inner();

    while let Some(event) = stream.next().await {
        match event {
            Ok(event) => {
                let _timestamp = &event.timestamp;

                if let Some(balance) = &event.balance_update {
                    println!("Balance updated:");
                    if let Some(currency) = &balance.currency {
                        println!("  Currency: {}", currency.asset_id);
                    }
                    if let Some(bal) = &balance.balance {
                        println!("  Available: {}", bal.sending);
                    }
                }

                if let Some(update) = &event.order_update {
                    if let Some(created) = &update.order_created {
                        println!("Order created: {}", created.order_id);
                    }

                    if let Some(completed) = &update.order_completed {
                        println!("Order completed: {}", completed.order_id);
                    }

                    if let Some(canceled) = &update.order_canceled {
                        println!("Order cancelled: {}", canceled.order_id);
                    }
                }

                if let Some(matched) = &event.order_matched {
                    println!("Order matched:");
                    println!("  Order ID: {}", matched.own_order_id);
                    println!("  Swap role: {:?}", matched.swap_role());
                    println!("  Routes: {}", matched.swap_route.len());
                }

                if let Some(swap) = &event.swap_update {
                    if let Some(progress) = &swap.progress {
                        println!("Swap update:");
                        println!("  Swap ID: {}", swap.swap_id);
                        println!("  Status: {:?}", progress.status);
                        println!("  Receiving: {}", progress.receiving_amount);
                        println!("  Paying: {}", progress.paying_amount);

                        if !progress.error.is_empty() {
                            println!("  Error: {}", progress.error);
                        }
                    }
                }

                if let Some(trade) = &event.market_trade_update {
                    println!("Trade executed: {:?}", trade.trade);
                }
            }
            Err(e) => {
                eprintln!("Stream error: {}", e);
                break;
            }
        }
    }

    println!("Stream ended");
    Ok(())
}

package main

import (
    "context"
    "fmt"
    "io"
    "log"

    pb "path/to/proto"
    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials/insecure"
)

func main() {
    conn, err := grpc.Dial("localhost:5003", grpc.WithTransportCredentials(insecure.NewCredentials()))
    if err != nil {
        log.Fatalf("Failed to connect: %v", err)
    }
    defer conn.Close()

    client := pb.NewSwapServiceClient(conn)

    request := &pb.SubscribeSimpleSwapsRequest{}
    stream, err := client.SubscribeSimpleSwaps(context.Background(), request)
    if err != nil {
        log.Fatalf("Failed to subscribe: %v", err)
    }

    for {
        update, err := stream.Recv()
        if err == io.EOF {
            fmt.Println("Stream ended")
            break
        }
        if err != nil {
            log.Fatalf("Stream error: %v", err)
        }

        swapId := update.GetSimpleSwapId()

        // Channel setup
        if funding := update.GetFundingSendingChannel(); funding != nil {
            fmt.Printf("[%s] Funding sending channel:\n", swapId)
            fmt.Printf("  Txid: %s\n", funding.GetTxid())
            fmt.Printf("  Channel ID: %s\n", funding.GetChannelId())
            fmt.Printf("  Amount: %s\n", funding.GetAmount())
            fmt.Printf("  Is Opening: %v\n", funding.GetIsOpening())
        }

        if leasing := update.GetLeasingReceivingChannel(); leasing != nil {
            fmt.Printf("[%s] Leasing receiving channel:\n", swapId)
            fmt.Printf("  Txid: %s\n", leasing.GetTxid())
            fmt.Printf("  Channel ID: %s\n", leasing.GetChannelId())
            fmt.Printf("  Amount: %s\n", leasing.GetAmount())
        }

        if dualFund := update.GetDualFundingChannel(); dualFund != nil {
            fmt.Printf("[%s] Dual-funding channel:\n", swapId)
            fmt.Printf("  Txid: %s\n", dualFund.GetTxid())
            fmt.Printf("  Self Amount: %s\n", dualFund.GetSelfAmount())
            fmt.Printf("  Counterparty Amount: %s\n", dualFund.GetCounterpartyAmount())
        }

        // Channel ready
        if update.GetSendingChannelReady() != nil {
            fmt.Printf("[%s] ✓ Sending channel ready\n", swapId)
        }

        if update.GetReceivingChannelReady() != nil {
            fmt.Printf("[%s] ✓ Receiving channel ready\n", swapId)
        }

        // Balance status
        if waiting := update.GetWaitingForBalances(); waiting != nil {
            fmt.Printf("[%s] Waiting for balances:\n", swapId)
            fmt.Printf("  Sending: %s\n", waiting.GetNeededSending())
            fmt.Printf("  Receiving: %s\n", waiting.GetNeededReceiving())
        }

        if update.GetBalancesReady() != nil {
            fmt.Printf("[%s] ✓ Balances ready\n", swapId)
        }

        // Order execution
        if order := update.GetOrderCreated(); order != nil {
            fmt.Printf("[%s] Order created: %s\n", swapId, order.GetOrderId())
        }

        if completed := update.GetOrderCompleted(); completed != nil {
            fmt.Printf("[%s] ✓ Order completed:\n", swapId)
            fmt.Printf("  Sent: %s\n", completed.GetSentAmount())
            fmt.Printf("  Received: %s\n", completed.GetReceivedAmount())
        }

        // Withdrawals
        if withdrawing := update.GetWithdrawingReceivingFunds(); withdrawing != nil {
            fmt.Printf("[%s] Withdrawing funds:\n", swapId)
            fmt.Printf("  Txids: %v\n", withdrawing.GetTxids())
            fmt.Printf("  Amount: %s\n", withdrawing.GetSelfAmount())
        }

        if update.GetReceivingFundsWithdrawn() != nil {
            fmt.Printf("[%s] ✓ Funds withdrawn\n", swapId)
        }

        // Final status
        if update.GetSimpleSwapCompleted() != nil {
            fmt.Printf("[%s] ✓✓✓ SWAP COMPLETED SUCCESSFULLY ✓✓✓\n", swapId)
        }

        if swapError := update.GetSimpleSwapError(); swapError != nil {
            fmt.Printf("[%s] ✗ ERROR: %s\n", swapId, swapError.GetError())
        }
    }
}

use futures::stream::StreamExt;
use proto::swap_service_client::SwapServiceClient;
use proto::SubscribeSimpleSwapsRequest;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut client = SwapServiceClient::connect("http://localhost:5003").await?;

    let request = SubscribeSimpleSwapsRequest {};
    let mut stream = client.subscribe_simple_swaps(request).await?.into_inner();

    while let Some(update) = stream.next().await {
        match update {
            Ok(update) => {
                let swap_id = &update.simple_swap_id;

                // Channel setup
                if let Some(funding) = &update.funding_sending_channel {
                    println!("[{}] Funding sending channel:", swap_id);
                    println!("  Txid: {}", funding.txid);
                    println!("  Channel ID: {}", funding.channel_id);
                    println!("  Amount: {}", funding.amount);
                    println!("  Is Opening: {}", funding.is_opening);
                }

                if let Some(leasing) = &update.leasing_receiving_channel {
                    println!("[{}] Leasing receiving channel:", swap_id);
                    println!("  Txid: {}", leasing.txid);
                    println!("  Channel ID: {}", leasing.channel_id);
                    println!("  Amount: {}", leasing.amount);
                }

                if let Some(dual_fund) = &update.dual_funding_channel {
                    println!("[{}] Dual-funding channel:", swap_id);
                    println!("  Txid: {}", dual_fund.txid);
                    println!("  Self Amount: {}", dual_fund.self_amount);
                    println!("  Counterparty Amount: {}", dual_fund.counterparty_amount);
                }

                // Channel ready
                if update.sending_channel_ready.is_some() {
                    println!("[{}] ✓ Sending channel ready", swap_id);
                }

                if update.receiving_channel_ready.is_some() {
                    println!("[{}] ✓ Receiving channel ready", swap_id);
                }

                // Balance status
                if let Some(waiting) = &update.waiting_for_balances {
                    println!("[{}] Waiting for balances:", swap_id);
                    println!("  Sending: {}", waiting.needed_sending);
                    println!("  Receiving: {}", waiting.needed_receiving);
                }

                if update.balances_ready.is_some() {
                    println!("[{}] ✓ Balances ready", swap_id);
                }

                // Order execution
                if let Some(order) = &update.order_created {
                    println!("[{}] Order created: {}", swap_id, order.order_id);
                }

                if let Some(completed) = &update.order_completed {
                    println!("[{}] ✓ Order completed:", swap_id);
                    println!("  Sent: {}", completed.sent_amount);
                    println!("  Received: {}", completed.received_amount);
                }

                // Withdrawals
                if let Some(withdrawing) = &update.withdrawing_receiving_funds {
                    println!("[{}] Withdrawing funds:", swap_id);
                    println!("  Txids: {:?}", withdrawing.txids);
                    println!("  Amount: {}", withdrawing.self_amount);
                }

                if update.receiving_funds_withdrawn.is_some() {
                    println!("[{}] ✓ Funds withdrawn", swap_id);
                }

                // Final status
                if update.simple_swap_completed.is_some() {
                    println!("[{}] ✓✓✓ SWAP COMPLETED SUCCESSFULLY ✓✓✓", swap_id);
                }

                if let Some(error) = &update.simple_swap_error {
                    eprintln!("[{}] ✗ ERROR: {}", swap_id, error.error);
                }
            }
            Err(e) => {
                eprintln!("Stream error: {}", e);
                break;
            }
        }
    }

    println!("Stream ended");
    Ok(())
}

func trackSwap(client pb.SwapServiceClient, swapId string) error {
    request := &pb.SubscribeSimpleSwapsRequest{}
    stream, err := client.SubscribeSimpleSwaps(context.Background(), request)
    if err != nil {
        return fmt.Errorf("failed to subscribe: %w", err)
    }

    for {
        update, err := stream.Recv()
        if err == io.EOF {
            return fmt.Errorf("stream ended unexpectedly")
        }
        if err != nil {
            return fmt.Errorf("stream error: %w", err)
        }

        // Filter for our swap
        if update.GetSimpleSwapId() != swapId {
            continue
        }

        if update.GetSimpleSwapCompleted() != nil {
            return nil // Success
        }

        if swapError := update.GetSimpleSwapError(); swapError != nil {
            return fmt.Errorf("swap error: %s", swapError.GetError())
        }
    }
}

async fn track_swap(
    client: &mut SwapServiceClient<tonic::transport::Channel>,
    swap_id: String,
) -> Result<(), Box<dyn std::error::Error>> {
    let request = SubscribeSimpleSwapsRequest {};
    let mut stream = client.subscribe_simple_swaps(request).await?.into_inner();

    while let Some(update) = stream.next().await {
        let update = update?;

        // Filter for our swap
        if update.simple_swap_id != swap_id {
            continue;
        }

        if update.simple_swap_completed.is_some() {
            return Ok(()); // Success
        }

        if let Some(error) = update.simple_swap_error {
            return Err(format!("swap error: {}", error.error).into());
        }
    }

    Err("stream ended unexpectedly".into())
}

var priceHistory []float64

stream, err := client.SubscribeMarketEvents(context.Background(), request)
if err != nil {
    log.Fatalf("Failed to subscribe: %v", err)
}

for {
    event, err := stream.Recv()
    if err == io.EOF {
        break
    }
    if err != nil {
        log.Fatalf("Stream error: %v", err)
    }

    if trade := event.GetTradeUpdate(); trade != nil {
        price, err := strconv.ParseFloat(trade.GetPrice(), 64)
        if err != nil {
            log.Printf("Failed to parse price: %v", err)
            continue
        }
        priceHistory = append(priceHistory, price)

        // Update chart
        updatePriceChart(priceHistory)
    }
}

let mut price_history: Vec<f64> = Vec::new();

let mut stream = client.subscribe_market_events(request).await?.into_inner();

while let Some(event) = stream.next().await {
    match event {
        Ok(event) => {
            if let Some(trade) = &event.trade_update {
                if let Ok(price) = trade.price.parse::<f64>() {
                    price_history.push(price);

                    // Update chart
                    update_price_chart(&price_history);
                }
            }
        }
        Err(e) => {
            eprintln!("Stream error: {}", e);
            break;
        }
    }
}

orderbookDepth := make(map[string]*pb.LiquidityPosition)

stream, err := client.SubscribeMarketEvents(context.Background(), request)
if err != nil {
    log.Fatalf("Failed to subscribe: %v", err)
}

for {
    event, err := stream.Recv()
    if err == io.EOF {
        break
    }
    if err != nil {
        log.Fatalf("Stream error: %v", err)
    }

    if update := event.GetOrderbookUpdate(); update != nil {
        // Add/update orders
        for orderId, position := range update.GetUpdatedOrders() {
            orderbookDepth[orderId] = position
        }

        // Remove orders
        for _, orderId := range update.GetRemovedOrders() {
            delete(orderbookDepth, orderId)
        }

        fmt.Printf("Current depth: %d orders\n", len(orderbookDepth))
    }
}

use std::collections::HashMap;

let mut orderbook_depth: HashMap<String, LiquidityPosition> = HashMap::new();

let mut stream = client.subscribe_market_events(request).await?.into_inner();

while let Some(event) = stream.next().await {
    match event {
        Ok(event) => {
            if let Some(update) = &event.orderbook_update {
                // Add/update orders
                for (order_id, position) in &update.updated_orders {
                    orderbook_depth.insert(order_id.clone(), position.clone());
                }

                // Remove orders
                for order_id in &update.removed_orders {
                    orderbook_depth.remove(order_id);
                }

                println!("Current depth: {} orders", orderbook_depth.len());
            }
        }
        Err(e) => {
            eprintln!("Stream error: {}", e);
            break;
        }
    }
}

// 1. Create subscription
stream, err := client.SubscribeMarketEvents(context.Background(), request)
if err != nil {
    log.Fatalf("Failed to subscribe: %v", err)
}

// 2. Listen to events
for {
    event, err := stream.Recv()
    if err == io.EOF {
        // Stream ended - reconnect if needed
        break
    }
    if err != nil {
        // Handle error
        log.Printf("Stream error: %v", err)
        break
    }

    // Handle event
    // ...
}

// 3. Stream automatically closes when loop exits
// For manual cancellation, use context.WithCancel

// 1. Create subscription
let mut stream = client.subscribe_market_events(request).await?.into_inner();

// 2. Listen to events
while let Some(event) = stream.next().await {
    match event {
        Ok(event) => {
            // Handle event
            // ...
        }
        Err(e) => {
            // Handle error
            eprintln!("Stream error: {}", e);
            break;
        }
    }
}

// 3. Stream automatically closes when dropped
// For manual cancellation, drop the stream or use tokio::select!