// Path: examples/src/main/java/fixio/examples/common/AbstractQuoteStreamingWorker.java
// public abstract class AbstractQuoteStreamingWorker implements Runnable {
// 
//     private static final int BUFFER_LENGTH = 32;
// 
//     private final BlockingQueue<Quote> quoteQueue;
// 
//     public AbstractQuoteStreamingWorker(BlockingQueue<Quote> quoteQueue) {
//         this.quoteQueue = quoteQueue;
//     }
// 
//     protected abstract void sendQuotes(List<Quote> buffer);
// 
//     @Override
//     public void run() {
//         final List<Quote> buffer = new ArrayList<>(BUFFER_LENGTH);
//         //noinspection InfiniteLoopStatement
//         while (true) {
//             quoteQueue.drainTo(buffer, BUFFER_LENGTH);
//             sendQuotes(buffer);
//             buffer.clear();
//         }
//     }
// 
// }
// 
// Path: examples/src/main/java/fixio/examples/common/Quote.java
// public class Quote {
// 
//     private final double bid;
//     private final double offer;
// 
//     public Quote(double bid, double offer) {
//         this.bid = bid;
//         this.offer = offer;
//     }
// 
//     public double getBid() {
//         return bid;
//     }
// 
//     public double getOffer() {
//         return offer;
//     }
// }


import fixio.examples.common.AbstractQuoteStreamingWorker;
import fixio.examples.common.Quote;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import quickfix.Application;
import quickfix.DoubleField;
import quickfix.FieldNotFound;
import quickfix.Message;
import quickfix.Session;
import quickfix.SessionID;
import quickfix.SessionNotFound;
import quickfix.field.MktBidPx;
import quickfix.field.MktOfferPx;
import quickfix.field.MsgType;
import quickfix.field.QuoteReqID;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
    @Override
    public void fromApp(Message message, SessionID sessionID) {
        try {
            String msgType = message.getHeader().getString(35);
            switch (msgType) {
                case MsgType.QUOTE_REQUEST:
                    String reqId = message.getString(QuoteReqID.FIELD);
                    subscriptions.put(reqId, sessionID);

                    LOGGER.debug("Subscribed with QuoteReqID={}", reqId);
                    break;
                case MsgType.QUOTE_CANCEL:
                    reqId = message.getString(QuoteReqID.FIELD);
                    subscriptions.remove(reqId);
                    LOGGER.debug("Unsubscribed with QuoteReqID={}", reqId);
                    break;

            }
        } catch (FieldNotFound fieldNotFound) {
            fieldNotFound.printStackTrace();
        }
    }

    private void stopStreaming(SessionID sessionID) {
        ArrayList<String> requestsToCancel = new ArrayList<>(subscriptions.size());
        requestsToCancel.addAll(subscriptions.entrySet().stream().filter(entry -> entry.getValue() == sessionID).map(Map.Entry::getKey).collect(Collectors.toList()));
        requestsToCancel.forEach(subscriptions::remove);
        LOGGER.info("Streaming Stopped for {}", sessionID);
    }

// Path: examples/src/main/java/fixio/examples/common/AbstractQuoteStreamingWorker.java
// public abstract class AbstractQuoteStreamingWorker implements Runnable {
// 
//     private static final int BUFFER_LENGTH = 32;
// 
//     private final BlockingQueue<Quote> quoteQueue;
// 
//     public AbstractQuoteStreamingWorker(BlockingQueue<Quote> quoteQueue) {
//         this.quoteQueue = quoteQueue;
//     }
// 
//     protected abstract void sendQuotes(List<Quote> buffer);
// 
//     @Override
//     public void run() {
//         final List<Quote> buffer = new ArrayList<>(BUFFER_LENGTH);
//         //noinspection InfiniteLoopStatement
//         while (true) {
//             quoteQueue.drainTo(buffer, BUFFER_LENGTH);
//             sendQuotes(buffer);
//             buffer.clear();
//         }
//     }
// 
// }
// 
// Path: examples/src/main/java/fixio/examples/common/Quote.java
// public class Quote {
// 
//     private final double bid;
//     private final double offer;
// 
//     public Quote(double bid, double offer) {
//         this.bid = bid;
//         this.offer = offer;
//     }
// 
//     public double getBid() {
//         return bid;
//     }
// 
//     public double getOffer() {
//         return offer;
//     }
// }


// Path: examples/src/main/java/fixio/examples/quickfix/QuickFixStreamingApp.java
import fixio.examples.common.AbstractQuoteStreamingWorker;
import fixio.examples.common.Quote;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import quickfix.Application;
import quickfix.DoubleField;
import quickfix.FieldNotFound;
import quickfix.Message;
import quickfix.Session;
import quickfix.SessionID;
import quickfix.SessionNotFound;
import quickfix.field.MktBidPx;
import quickfix.field.MktOfferPx;
import quickfix.field.MsgType;
import quickfix.field.QuoteReqID;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

    @Override
    public void fromApp(Message message, SessionID sessionID) {
        try {
            String msgType = message.getHeader().getString(35);
            switch (msgType) {
                case MsgType.QUOTE_REQUEST:
                    String reqId = message.getString(QuoteReqID.FIELD);
                    subscriptions.put(reqId, sessionID);

                    LOGGER.debug("Subscribed with QuoteReqID={}", reqId);
                    break;
                case MsgType.QUOTE_CANCEL:
                    reqId = message.getString(QuoteReqID.FIELD);
                    subscriptions.remove(reqId);
                    LOGGER.debug("Unsubscribed with QuoteReqID={}", reqId);
                    break;

            }
        } catch (FieldNotFound fieldNotFound) {
            fieldNotFound.printStackTrace();
        }
    }

    private void stopStreaming(SessionID sessionID) {
        ArrayList<String> requestsToCancel = new ArrayList<>(subscriptions.size());
        requestsToCancel.addAll(subscriptions.entrySet().stream().filter(entry -> entry.getValue() == sessionID).map(Map.Entry::getKey).collect(Collectors.toList()));
        requestsToCancel.forEach(subscriptions::remove);
        LOGGER.info("Streaming Stopped for {}", sessionID);
    }

    private class StreamingWorker extends AbstractQuoteStreamingWorker {