Connect to Cardano node through Yaci reactive api and aggregations for 5 min windows

pom.xml

    <dependencies>
        <dependency>
            <groupId>com.bloxbean.cardano</groupId>
            <artifactId>yaci-core</artifactId>
            <version>0.0.8</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.7.36</version>
        </dependency>
    </dependencies>

ReactiveWindow.java

import com.bloxbean.cardano.yaci.core.common.NetworkType;
import com.bloxbean.cardano.yaci.core.reactive.BlockStreamer;

import java.text.DecimalFormat;
import java.time.Duration;
import java.util.Date;
import java.util.TimeZone;

/**
 * Get incoming block data through reactive BlockStreamer api
 */
public class ReactiveWindow {
    private final long MAX_BLOCK_BODY_SIZE = 90112;
    private final long TIME_WINDOW_IN_MINS = 5;

    public void start() {
        DecimalFormat df = new DecimalFormat("0.00");

        BlockStreamer.fromLatest(NetworkType.MAINNET).stream()
                .window(Duration.ofMinutes(TIME_WINDOW_IN_MINS))
                .doOnNext(blockFlux -> {
                    blockFlux.map(block -> {
                                System.out.println("-------------------");
                                System.out.println("Blk # : " + block.getHeader().getHeaderBody().getBlockNumber() + ",  "
                                        + "Size : " + df.format(block.getHeader().getHeaderBody().getBlockBodySize() / 1024.0) + " KB , "
                                        + " # of Txs : " + block.getTransactionBodies().size());
                                return block;
                            }).collectList()
                            .subscribe(blocks -> {
                                Long totalFee = blocks.stream()
                                        .flatMap(block -> block.getTransactionBodies().stream()
                                                .map(transactionBody -> transactionBody.getFee().longValue()))
                                        .reduce(0L, Long::sum);

                                Long totalSize = blocks.stream()
                                        .map(block -> block.getHeader().getHeaderBody().getBlockBodySize())
                                        .reduce(0L, Long::sum);

                                int totalMint = blocks.stream()
                                        .flatMap(block -> block.getTransactionBodies().stream()
                                                .map(transactionBody -> transactionBody.getMint().size()))
                                        .reduce(0, Integer::sum);

                                Long totalTxns = blocks.stream().map(block -> block.getTransactionBodies().size())
                                        .reduce(0, Integer::sum)
                                        .longValue();

                                double tps = totalTxns / (double)(TIME_WINDOW_IN_MINS * 60);

                                int totalOutputs = blocks.stream()
                                        .flatMap(block -> block.getTransactionBodies().stream()
                                                .map(transactionBody -> transactionBody.getOutputs().size()))
                                        .reduce(0, Integer::sum);
                                double tops = totalOutputs / (double)(TIME_WINDOW_IN_MINS * 60);

                                double blkSizePerct = ((totalSize / (double)blocks.size()) / MAX_BLOCK_BODY_SIZE ) * 100;
                                double avgBlockSize = blocks.size() != 0? (totalSize / (double)blocks.size()) / 1024.0: 0;

                                System.out.println("#################################################");
                                System.out.println("Time : " + new Date());
                                System.out.println("Total Fee: " + totalFee / 1000000.0 + " Ada");
                                System.out.println("\nTotal blockSize (KB): " + totalSize / 1024.0);
                                System.out.println("Avg block size: " + df.format(avgBlockSize) + " KB" + " , avg % : " + df.format(blkSizePerct) + "%");
                                System.out.println("\n# of New Mints: " + totalMint);
                                System.out.println("Total # of tnxns: " + totalTxns);
                                System.out.println("Total # of txn outputs: " + totalOutputs);

                                System.out.println("Txns / sec : " + df.format(tps));
                                System.out.println("Txn Outputs / sec : " + df.format(tops));
                                System.out.println("#################################################");
                            });
                })
                .subscribe();
    }

    public static void main(String[] args) {
        new ReactiveWindow().start();
    }
}