ExchangePipeline.java

package concurrency.module06;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Conceptual exchange pipeline: orders enter an ingest queue; a single matching
 * thread consumes from it, runs matching via {@link OrderBook}, and pushes fills
 * to a fill queue. Downstream (risk, settlement, notifications) would consume
 * from the fill queue. This separation keeps matching single-threaded and allows
 * other stages to scale independently.
 */
public class ExchangePipeline {

    /** Order as submitted from the gateway; converted to OrderBook.Order for matching. */
    static class IncomingOrder {
        final long id;
        final boolean buy;
        final double price;
        final int quantity;
        final String userId;

        IncomingOrder(long id, boolean buy, double price, int quantity, String userId) {
            this.id = id;
            this.buy = buy;
            this.price = price;
            this.quantity = quantity;
            this.userId = userId;
        }
    }

    /** A fill (trade): order id, price, quantity. Emitted to fill queue for risk/settlement. */
    static class Fill {
        final long orderId;
        final double price;
        final int quantity;

        Fill(long orderId, double price, int quantity) {
            this.orderId = orderId;
            this.price = price;
            this.quantity = quantity;
        }
    }

    private final BlockingQueue<IncomingOrder> ingestQueue = new LinkedBlockingQueue<>();
    private final BlockingQueue<Fill> fillQueue = new LinkedBlockingQueue<>();
    private final OrderBook book;
    private final AtomicLong orderIdGen = new AtomicLong(0);
    private volatile boolean running = true;
    private final Thread matchingThread;

    /**
     * Starts the matching thread: poll ingest queue, match each order against the book,
     * push fills to fill queue. Single-threaded matching ensures order book consistency.
     *
     * @param symbol symbol for the order book (e.g. "BTC-USD")
     */
    public ExchangePipeline(String symbol) {
        this.book = new OrderBook(symbol);
        matchingThread = new Thread(() -> {
            while (running) {
                try {
                    IncomingOrder in = ingestQueue.poll(100, TimeUnit.MILLISECONDS);
                    if (in != null) {
                        OrderBook.Order order = new OrderBook.Order(in.id, in.buy, in.price, in.quantity, in.userId);
                        for (double[] f : book.match(order)) {
                            fillQueue.offer(new Fill(in.id, f[0], (int) f[1]));
                        }
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        });
        matchingThread.start();
    }

    /**
     * Submits an order to the pipeline. Returns immediately with an order id;
     * matching runs asynchronously on the matching thread.
     *
     * @return assigned order id
     */
    public long submitOrder(boolean buy, double price, int quantity, String userId) {
        long id = orderIdGen.incrementAndGet();
        ingestQueue.offer(new IncomingOrder(id, buy, price, quantity, userId));
        return id;
    }

    /** Returns the fill queue so downstream (risk, settlement) can consume fills. */
    public BlockingQueue<Fill> getFillQueue() {
        return fillQueue;
    }

    public void stop() {
        running = false;
        matchingThread.interrupt();
    }
}