add new api for passing protobuf quote

NodeActiveTick.cpp

@@ -48,6 +48,7 @@ NAN_MODULE_INIT(NodeActiveTick::Init) {
     Nan::SetPrototypeMethod(tpl, "quoteStreamRequest", QuoteStreamRequest);
     Nan::SetPrototypeMethod(tpl, "barHistoryDbRequest", BarHistoryDbRequest);
     Nan::SetPrototypeMethod(tpl, "quoteDbRequest", QuoteDbRequest);
+    Nan::SetPrototypeMethod(tpl, "quoteStreamRequestForSymbolData", QuoteStreamRequestForSymbolData);
 
     // constructor.Reset(isolate, tpl->GetFunction());
     constructor().Reset(Nan::GetFunction(tpl).ToLocalChecked());
@@ -238,6 +239,34 @@ NAN_METHOD(NodeActiveTick::ListRequest) {
   info.GetReturnValue().Set(Nan::New<Number>(s_pInstance->m_hLastRequest));
 }
 
+NAN_METHOD(NodeActiveTick::QuoteStreamRequestForSymbolData) {
+  uint64_t quote_stream_request = 0;
+  NodeActiveTick *obj = ObjectWrap::Unwrap<NodeActiveTick>(info.Holder());
+  // void* buf = (void*) node::Buffer::Data(info[0]->ToObject());
+  // Nan::Utf8String p_data(info[2]->ToString());
+  // std::string data_string = std::string(*p_data);
+  char* buf = node::Buffer::Data(info[0]->ToObject());
+
+  unsigned int size = info[1]->Uint32Value();
+  Nan::Utf8String request_type(info[2]->ToString());
+  std::string str_request_type = std::string(*request_type);
+
+  ATStreamRequestType requestType = obj->enumConverter->toAtStreamRequest(str_request_type);  
+  NodeActiveTickProto::ATSymbol s = NodeActiveTickProto::ATSymbol();
+
+  s.ParseFromArray(buf, size);
+  ATSYMBOL atSymbol;
+  // atSymbol.symbol = s.symbol();
+  Helper::ConvertString(s.symbol().c_str(), atSymbol.symbol, ATSymbolMaxLength);
+  atSymbol.symbolType = s.symboltype();
+  atSymbol.exchangeType = s.exchangetype();
+  atSymbol.countryType = s.countrytype();
+  std::vector<ATSYMBOL> out;
+  out.push_back(atSymbol);
+  quote_stream_request = obj->requestor->SendATQuoteStreamRequest(out.data(), 1, requestType, DEFAULT_REQUEST_TIMEOUT);
+  info.GetReturnValue().Set(Nan::New<Number>(quote_stream_request));
+}
+
 NAN_METHOD(NodeActiveTick::QuoteStreamRequest) {
   uint64_t quote_stream_request = 0;
   NodeActiveTick *obj = ObjectWrap::Unwrap<NodeActiveTick>(info.Holder());

NodeActiveTick.h

@@ -57,6 +57,7 @@ class NodeActiveTick : public Nan::ObjectWrap {
     static NAN_METHOD(QuoteStreamRequest);
     static NAN_METHOD(BarHistoryDbRequest);
     static NAN_METHOD(QuoteDbRequest);
+    static NAN_METHOD(QuoteStreamRequestForSymbolData);
     // AT Callbacks
     static void ATStreamUpdateCallback(LPATSTREAM_UPDATE pUpdate);
     static void ATSessionStatusChangeCallback(

iced/ActiveTick.iced

@@ -34,6 +34,7 @@ class ActiveTick extends EventEmitter
       @ATQuoteStreamQuoteUpdate = @messages_builder.build 'NodeActiveTickProto.ATQuoteStreamQuoteUpdate'
       @ATBarHistoryDbResponse = @messages_builder.build 'NodeActiveTickProto.ATBarHistoryDbResponse'
       @ATQuoteDbResponse = @messages_builder.build 'NodeActiveTickProto.ATQuoteDbResponse'
+      @ATSymbol = @messages_builder.build 'NodeActiveTickProto.ATSymbol'
 
       # enums
       @ATQuoteFieldTypes = @messages_builder.build 'NodeActiveTickProto.ATQuoteFieldType'
@@ -43,6 +44,9 @@ class ActiveTick extends EventEmitter
       @ATSymbolStatus = @messages_builder.build 'NodeActiveTickProto.ATSymbolStatus'
       @ATQuoteFieldType = @messages_builder.build 'NodeActiveTickProto.ATQuoteFieldType'
       @ATStreamRequestTypes = @messages_builder.build 'NodeActiveTickProto.ATStreamRequestTypes'
+
+
+      @stream_symbols = {}
       readyCb()
 
   barHistoryDBRequest: (symbol, barhistorytype, intradayminutecompression, startime, endtime, requestCb) =>
@@ -53,20 +57,53 @@ class ActiveTick extends EventEmitter
     request_id = @api.quoteDbRequest symbol, fields
     @callbacks[request_id] = requestCb if requestCb?
 
-  quoteStreamRequest: (symbols, request_type, requestCb) =>
-    if typeof symbols is 'object'
-      if symbols.length is 1
-        symbolParam = symbols[0]
-        symbolCount = 1
+  # subscribe/unsubscribe prevent duplicating server requests if many clients subscribe to similar symbols
+  subscribe: (symbol, cb) =>
+    return if not symbol?
+    @stream_symbols[symbol] = 0 if not @stream_symbols[symbol]?
+    if @stream_symbols[symbol] is 0
+      @quoteStreamRequest symbol, 'StreamRequestSubscribe', (msg) =>
+        console.log msg
+        @stream_symbols[symbol] += 1
+        cb yes
+    else if @stream_symbols[symbol] > 0 # we're already getting this quote stream
+      @stream_symbols[symbol] += 1
+      cb yes
+
+  unsubscribe: (symbol, cb) =>
+    return if not symbol?
+    if @stream_symbols[symbol] is 0 or not @stream_symbols[symbol]?
+      return console.error "Unsubscribe sent for symbol we aren't subscribed to #{symbol}"
+    else
+      @stream_symbols[symbol] -= 1
+      if @stream_symbols[symbol] <= 0
+        @quoteStreamRequest symbol, 'StreamRequestUnsubscribe', (msg) =>
+          console.log msg
+          cb yes
       else
-        symbolParam = symbols.join ','
-        symbolCount = symbols.length
-    else if typeof symbols is 'string'
-      symbolParam = symbols
+        cb yes
+
+  quoteStreamRequest: (symbol, reqAction, requestCb) =>
+    # reqAction = 'StreamRequestSubscribe' # determined
+    if typeof symbol is 'object'
+      console.log 'object'
+      q = new @ATSymbol
+      q.symbol = symbol.symbol.toString('utf8')
+      q.symbolType = symbol.symbolType
+      q.exchangeType = symbol.exchangeType
+      q.countryType = symbol.countryType
+      r = q.encode()
+      request_id = @api.quoteStreamRequestForSymbolData r.buffer, r.buffer.length, reqAction
+      @callbacks[request_id] = requestCb if requestCb?
+    else if typeof symbol is 'string'
+      console.log 'string'
+      symbolParam = symbol
       symbolCount = 1
-    request_id = @api.quoteStreamRequest symbolParam, symbolCount, request_type
-    @callbacks[request_id] = requestCb if requestCb?
-
+      request_id = @api.quoteStreamRequest symbolParam, symbolCount, reqAction
+      @callbacks[request_id] = requestCb if requestCb?
+    else
+      console.error 'Bad request parameter type quoteStreamRequest'
+
   # "ATConstituentListIndex"
   # "ATConstituentListSector"
   # "ATConstituentListOptionChain"
@@ -90,9 +127,11 @@ class ActiveTick extends EventEmitter
       msg = @ATBarHistoryDbResponse.decode msgData
     else if msgType is 'ATQuoteStreamTradeUpdate'
       msg = @ATQuoteStreamTradeUpdate.decode msgData
+      console.log 'trade'
       @emit 'trade', msg
     else if msgType is 'ATQuoteStreamQuoteUpdate'
       msg = @ATQuoteStreamQuoteUpdate.decode msgData
+      console.log 'quote'
       @emit 'quote', msg
     else if msgType is 'ATQuoteDbResponse'
       msg = @ATQuoteDbResponse.decode msgData