Event Subscriber Example

use clap::Parser;
use dotenv;
use drift_rs::event_subscriber::{DriftEvent, EventSubscriber};
use drift_rs::{PubsubClient, constants::PROGRAM_ID};
use env_logger;
use futures_util::StreamExt;
use std::env;
use std::sync::Arc;
use tokio;

#[derive(Parser, Debug)]
#[clap(author, version, about, long_about = None)]
struct Args {
    #[clap(long, action, help = "Use gRPC for event subscription")]
    grpc: bool,
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let _ = env_logger::init();
    let _ = dotenv::dotenv().ok();
    let args = Args::parse();

    let mut event_subscriber = if args.grpc {
        let grpc_endpoint = env::var("GRPC_ENDPOINT")
            .expect("GRPC_ENDPOINT must be set");
        let grpc_x_token = env::var("GRPC_X_TOKEN")
            .expect("GRPC_X_TOKEN must be set");
        EventSubscriber::subscribe_grpc(
            grpc_endpoint, 
            grpc_x_token, 
            PROGRAM_ID
        )
            .await
            .expect("error subscribing to grpc events")
    } else {
        let ws_rpc_endpoint = env::var("WS_RPC_ENDPOINT")
            .expect("WS_RPC_ENDPOINT must be set");
        let client = PubsubClient::new(&ws_rpc_endpoint)
            .await
            .unwrap();
        EventSubscriber::subscribe(Arc::new(client), PROGRAM_ID)
            .await
            .expect("error subscribing to ws events")
    };

    println!("subscribed to events");
    let mut count = 0;
    
    while let Some(event) = event_subscriber.next().await {
        match event {
            DriftEvent::OrderFill {
                maker,
                maker_order_id,
                maker_side,
                taker,
                taker_order_id,
                taker_side,
                market_index,
                market_type,
                signature,
                bit_flags,
                ..
            } => {
                println!(
                    "order fill: market: {}-{} maker: {}-{}-{} taker: {}-{}-{} signature: {} bit_flags: {}",
                    market_type.as_str(),
                    market_index,
                    maker.unwrap_or_default().to_string().as_str(),
                    maker_order_id,
                    format!("{:?}", maker_side.unwrap_or_default()),
                    taker.unwrap_or_default().to_string().as_str(),
                    taker_order_id,
                    format!("{:?}", taker_side.unwrap_or_default()),
                    signature,
                    bit_flags
                );
                count += 1;
                if count > 100 {
                    break;
                }
            }
            _ => {}
        }
    }

    event_subscriber.unsubscribe();
    Ok(())
}

DriftEvent::OrderFill {
    maker,              // Maker's public key
    maker_order_id,     // Maker's order ID
    maker_side,         // Buy or Sell
    taker,              // Taker's public key
    taker_order_id,     // Taker's order ID
    taker_side,         // Buy or Sell
    base_asset_amount_filled,   // Amount filled
    quote_asset_amount_filled,  // Quote value
    market_index,       // Market identifier
    market_type,        // Perp or Spot
    oracle_price,       // Oracle price at fill
    signature,          // Transaction signature
    ts,                 // Timestamp
    ..
}

DriftEvent::OrderCancel {
    order_id,          // The cancelled order ID
    user,              // User public key
    signature,         // Transaction signature
    ..
}

DriftEvent::OrderCreate {
    order,             // Full order details
    user,              // User public key
    ts,                // Timestamp
    signature,         // Transaction signature
    ..
}

DriftEvent::FundingPayment {
    amount,            // Funding amount
    market_index,      // Market identifier
    user,              // User public key
    ts,                // Timestamp
    ..
}

let ws_rpc_endpoint = "wss://api.mainnet-beta.solana.com";
let client = PubsubClient::new(&ws_rpc_endpoint).await?;
let mut subscriber = EventSubscriber::subscribe(
    Arc::new(client), 
    PROGRAM_ID
).await?;

let grpc_endpoint = "your-grpc-endpoint";
let grpc_token = "your-token";
let mut subscriber = EventSubscriber::subscribe_grpc(
    grpc_endpoint,
    grpc_token,
    PROGRAM_ID
).await?;

while let Some(event) = subscriber.next().await {
    match event {
        DriftEvent::OrderFill { .. } => {
            // Handle order fill
        },
        DriftEvent::OrderCancel { .. } => {
            // Handle order cancellation
        },
        _ => {
            // Ignore other events
        }
    }
}

# Set environment variables
export WS_RPC_ENDPOINT="wss://api.mainnet-beta.solana.com"

# Run
cargo run --example event-subscriber

# Set environment variables
export GRPC_ENDPOINT="your-grpc-endpoint"
export GRPC_X_TOKEN="your-token"

# Run with --grpc flag
cargo run --example event-subscriber -- --grpc

DriftEvent::OrderFill { maker, taker, .. } => {
    if maker == my_pubkey {
        // My maker order was filled
        update_inventory();
        adjust_quotes();
    }
}

DriftEvent::OrderFill { 
    market_index,
    base_asset_amount_filled,
    quote_asset_amount_filled,
    ..
} => {
    let price = quote_asset_amount_filled / base_asset_amount_filled;
    record_trade(market_index, price, base_asset_amount_filled);
}

DriftEvent::OrderFill { market_index, .. } => {
    if market_index == 0 { // SOL-PERP
        alert_team("SOL-PERP fill detected");
    }
}

use tokio::sync::mpsc;

// Create channel for events
let (tx, mut rx) = mpsc::channel(1000);

// Spawn event processor
tokio::spawn(async move {
    while let Some(event) = rx.recv().await {
        // Process event in background
        process_event(event).await;
    }
});

// Stream events to channel
while let Some(event) = subscriber.next().await {
    tx.send(event).await.ok();
}