stream eth2 blocks (and txs) directly from the consensus client [using checkpoint sync it takes a few minutes to receive the newest blocks]

block_stream.rs

// steps: 
// 1) setup and run lighthouse to run with checkpoint sync 
// 2) run akula execution client 
// 3) wait till lighthouse is 'synced' (takes a few minutes)
// 4) run this script to query the newest block + its txs 

use std::{result::Result, error::Error};
use hyper::{Client, body::HttpBody, Request};
use hyper::http::header::ACCEPT;
use rlp::{Rlp, Decodable}; 
use ethers::core::types::Transaction as EthersTransaction;

// lighthouse/consensus/types
use types::{SignedBeaconBlock, ChainSpec, MainnetEthSpec, FullPayload, Transactions}; 

#[tokio::main]
pub async fn main() -> Result<(), Box<dyn Error>> { 

    // query lighthouse http api for block
    let client = Client::new(); 
    let uri = "http://localhost:5052/eth/v2/beacon/blocks/head";

    let ssz_header = hyper::header::HeaderValue::from_static("application/octet-stream");
    let request = Request::builder()
        .method("GET")
        .uri(uri)
        .header(ACCEPT, ssz_header)
        .body(hyper::Body::empty())
        .unwrap();
    let mut resp = client.request(request).await?;

    // receive response 
    let mut bytes = vec![];
    while let Some(chunk) = resp.body_mut().data().await {
        let chunk = chunk?;
        bytes.extend_from_slice(&chunk);
    }
    let bytes = &bytes[..];

    // decode block 
    let spec = ChainSpec::mainnet();
    let block: SignedBeaconBlock<MainnetEthSpec, FullPayload<_>> = SignedBeaconBlock::from_ssz_bytes(bytes, &spec).unwrap();
    println!("found block @ slot: {:?}", block.message().slot());

    // decode transactions using RLP 
    let payload = block.message().body().execution_payload().unwrap();
    let payload = payload.execution_payload.clone();

    let txs: Transactions<MainnetEthSpec> = payload.transactions.clone();
    let vec_txs: Vec<_> = Vec::from(txs);
    println!("found {} txs", vec_txs.len());

    vec_txs.iter().for_each(|tx| {
        let tx: Vec<u8> = Vec::from(tx.clone());
        // decode using rlp 
        let rlp_tx = Rlp::new(&tx[..]);
        let tx = EthersTransaction::decode(&rlp_tx).expect("rlp decode failed: {tx:#?}");
        println!("found tx with 'to address': {:?}", tx.to);
    });

    Ok(())
}