how to implement subscribing to data feed? notify objects

Question

this question is aimed to those of us who really faced with this problem. I am interested in real, working, fast solutions.

Description:

I am using API which allows me to communicate over tcp/ip via Posix conformed socket to get live streaming data. These are prices of financial instruments and other their statistics, all being numbers. I have one Posix client that manages this connection via socket, this is Qt application. Each form (GUI) can request data stream and display incoming quotes.

Problem:

every incoming stream is received via a callback in socket (now I just print info)

void PosixClient::tickPrice( TickerId tickerId, TickType field, double price, int canAutoExecute) {
    printf("tradingclient_1: tickPrice: \n");}

The model concept introduced via this API is that streams are recognized due to tickerId field, so when new data appears on the socket the tickPrice method is fired and I should assign/notify etc interested objects about it - i.e. send data to appropriate GUI form differentiate them by tickerId.

Question:

how should I implement data exchange, subscription model to send data to correct objects? my first thought is to use in Posix client

std::vector<int,my_function> v;

which can map tickerId to callback from object which requested data. Something like Observer Pattern.

---

So at the moment I have an Observer Pattern implementation. This is main idea how it works:

Observable:

#include <cstdlib>
#include <ql/patterns/../patterns/observable.hpp>
#include <iostream> 
/*
 * 
 */

class DataRepository : public QuantLib::Observable{
    public:
        void run();
        int getData();
    private:
        int data;         
};

void DataRepository::run(){
    for(int i=0;i<10;++i){
        data=i;
        notifyObservers();
    }
}
int DataRepository::getData(){
    return data;
}

Observer:

typedef boost::shared_ptr<DataRepository> pMyObservable;
class myObserver : public QuantLib::Observer{
    public:
       myObserver(pMyObservable obs, std::string n)
        : observable(obs), name(n){
       this->registerWith(observable);
       }
       myObserver(const myObserver &observer)
       : Observer(observer),
        observable(observer.observable),
        name("Copy of "+observer.name){
       }

       void update(){
          data=observable->getData();
          std::cout<<"new data: "<<data<<std::endl;
       }
    private:
       int data;
       pMyObservable observable;
       std::string name;
};

example:

int main(int argc, char** argv) {

    pMyObservable d(new DataRepository);
    myObserver obs(d,"count_to_10_data");
    d->run();

    return 0;
}

result:

new data: 0

new data: 1

new data: 2

new data: 3

new data: 4

new data: 5

new data: 6

new data: 7

new data: 8

new data: 9

RUN SUCCESSFUL (total time: 93ms)

---

and here is my real code for this moment:

#include <ql/patterns/observable.hpp>
#include "Contract.h"
#include <boost/function.hpp>
#include "IB_events.h"
#include <list>

typedef boost::shared_ptr<IB::Record> rec_ptr;
typedef boost::shared_ptr<IB::TickPriceRecord> tickPriceRec_ptr;
typedef boost::shared_ptr<IB::TickSizeRecord>  tickSizeRec_ptr;
typedef boost::shared_ptr<IB::TickStringRecord>  tickStringRec_ptr;

class MarketData : public QuantLib::Observable {
public:
    MarketData();
    MarketData(IB::Event processedEvent, int tickerId, IB::Contract contractDescription):
    processedEvent(processedEvent), tickerId(tickerId), contractDescription(contractDescription) {}
    virtual ~MarketData();
    int getTickerId(){ return tickerId; }
    void putRecord(boost::shared_ptr<IB::Record> record){
        record_=record;
    }
    boost::shared_ptr<IB::Record> getRecord(){
        return record_;
    }
    IB::Event getEvent(){
        return processedEvent;
    }    
private:
    MarketData(const MarketData& orig);
    boost::shared_ptr<IB::Record> record_;
    // this MarketData object can handle these events
    // any observer can subscribe to one of those events
    IB::Event processedEvent;
    int tickerId;
    IB::Contract contractDescription;
};

further:

typedef boost::shared_ptr<MarketData> pMyObservable;
typedef boost::function<void (int tickerId, boost::shared_ptr<IB::Record> record)> f_action_ptr;

// one MarketDataObserver may observe one tickerId and for one event
// if you want to be notified when many events happened (i.e. TickSize and TickPrice)
// you can subscribe many MarketDataObservers to one and the same MarketData instance
class MarketDataObserver : public QuantLib::Observer{
public:
    MarketDataObserver(pMyObservable obs, IB::Event observedEvent, f_action_ptr ptr)
        : observable(obs), observedEvent_(observedEvent), f_ptr(ptr){
      this->registerWith(observable);
    }
    MarketDataObserver(const MarketDataObserver &observer)
      : Observer(observer),
        observable(observer.observable){ // faction_ptr is not copied!
    }

    // object which subscribed to data stream using this MarketDataObserver
    // will be notified about incoming IB::Record
    void update() {
        if (observable->getEvent() == observedEvent_) { // just to be 100% sure
            boost::shared_ptr<IB::Record> data = observable->getRecord();

            // here appropriate function is called: myTickPriceUpdate,
            // myTickSizeUpdate or myTickStringUpdate depending on what
            // subscribing object specified in f_action_ptr ptr
            // in MarketDataObserver constructor
            f_ptr(observable->getTickerId(), data);
        }
    }
private:
    pMyObservable observable;
    f_action_ptr f_ptr;
    IB::Event observedEvent_; // the single event in which observer is interested
};

typedef boost::shared_ptr<MarketData>  mktData_ptr;

But this has many disadvantages. Is there any better/cheaper/faster approach? What can I improve?

Answer 1

Yes, it's basically Observer pattern, but you need a neater subscription model.

Let's say each form implements an Observer interface containing a single method dataChanged(TickerId) which is called when a data for this tickerId gets updated.

If number of GUI forms is not very large, and it's acceptable to notify them all at every change, I'd suggest the following simple solution: the "subject" (PosixClient) maintains a list (maybe std::vector) of subscribed forms. When a callback from socket is fired, the client notifies ALL the forms, and those forms which are interested in the TickerId update their data. Yes, that's not optimal as most of the forms do not want the notification, but the overhead is usually not noticeable.

If there are too many GUI forms to notify them all at once (although I cannot imagine such a GUI), the PosixClient can maintain something like a std::multimap with TickerId as key and Observer* as value, so it can notify only those who subscribed to changes about given tickerId.

And I'd suggest to store all data (prices etc.) in some common object which can be queried by GUI forms rather than sending updated values along with dataChanged call.