Enable quotesource server

app/Main.hs

@@ -12,11 +12,13 @@ import Control.Concurrent.BoundedChan
 import Data.ATrade
 import QuoteSource.TableParsers.AllParamsTableParser
 import QuoteSource.TableParser
+import QuoteSource.Server
 
 import System.Log.Logger
 import System.Log.Handler.Simple
 import System.Log.Handler (setFormatter)
 import System.Log.Formatter
+import System.ZMQ4
 
 import Data.Aeson
 import Data.Aeson.Types
@@ -78,11 +80,11 @@ main = do
   config <- readConfig "quik-connector.config.json"  
   infoM "main" "Config loaded"
   chan <- newBoundedChan 1000
-  forkIO $ forever $ do
-    tick <- readChan chan
-    when (datatype tick == Price) $ print tick
   infoM "main" "Starting data import server"
   dis <- initDataImportServer [MkTableParser $ mkAllParamsTableParser "allparams"] chan "atrade"
+  withContext (\ctx -> do
+    qsServer <- startQuoteSourceServer chan ctx (quotesourceEndpoint config)
+
     void initGUI
     window <- windowNew
     window `on` deleteEvent $ do
@@ -90,4 +92,6 @@ main = do
       return False
     widgetShowAll window
     mainGUI
+    stopQuoteSourceServer qsServer
+    infoM "main" "Main thread done")
 

quik-connector.cabal

@@ -20,6 +20,7 @@ library
                      , Data.ATrade
                      , QuoteSource.TableParser
                      , QuoteSource.TableParsers.AllParamsTableParser
+                     , QuoteSource.Server
   ghc-options:         -Wincomplete-patterns
   build-depends:       base >= 4.7 && < 5
                      , Win32
@@ -57,6 +58,7 @@ executable quik-connector-exe
                      , unordered-containers
                      , vector
                      , text
+                     , zeromq4-haskell
   default-language:    Haskell2010
   extra-libraries:     "user32"
 

src/Broker.hs

@@ -0,0 +1,50 @@
+
+module Broker (
+) where
+
+import Data.Decimal
+import Data.Time.Clock
+
+data SignalId = SignalId {
+  strategyId :: String,
+  signalName :: String,
+  comment :: String }
+  deriving (Show, Eq)
+
+data OrderPrice = Market | Limit Decimal | Stop Decimal Decimal
+
+data Operation = Buy | Sell
+  deriving (Show, Eq)
+
+data OrderState = Unsubmitted
+  | Submitted
+  | PartiallyExecuted
+  | Executed
+  | Cancelled
+  | Rejected String
+  | Error String
+
+data Order = Order {
+  orderId :: Integer,
+  orderAccountId :: String,
+  orderSecurity :: String,
+  orderPrice :: OrderPrice,
+  orderQuantity :: Integer,
+  orderExecutedQuantity :: Integer,
+  orderOperation :: Operation,
+  orderState :: OrderState,
+  orderSignalId :: SignalId }
+  deriving (Show, Eq)
+
+data Trade = Trade {
+  tradeOrderId :: Integer,
+  tradePrice :: Decimal,
+  tradeQuantity :: Integer,
+  tradeVolume :: Decimal,
+  tradeVolumeCurrency :: String,
+  tradeAccount :: String,
+  tradeSecurity :: String,
+  tradeTimestamp :: UTCTime,
+  tradeSignalId :: SignalId }
+  deriving (Show, Eq)
+

src/QuoteSource/Server.hs

@@ -1,28 +1,36 @@
 
 module QuoteSource.Server (
+  startQuoteSourceServer,
+  stopQuoteSourceServer
 ) where
 
 import System.ZMQ4
 import Control.Concurrent.BoundedChan
 import Data.ATrade
-import Control.Concurrent
+import Control.Concurrent hiding (readChan)
 import Control.Monad
+import Control.Exception
+import qualified Data.ByteString.Lazy as BL
+import Data.List.NonEmpty hiding (map)
+import System.Log.Logger
 
 data QuoteSourceServer = QuoteSourceServerState {
   ctx :: Context,
   outSocket :: Socket Pub,
   tickChannel :: BoundedChan Tick,
-  serverThread :: ThreadId
+  serverThreadId :: ThreadId
 }
 
 serverThread :: QuoteSourceServer -> IO ()
-serverThread state = finally serverThread' cleanup
+serverThread state = do
+  finally serverThread' cleanup
+  debugM "QuoteSource" "server thread done"
   where
     cleanup = close $ outSocket state
 
-    serverThread = forever $ do
+    serverThread' = forever $ do
       tick <- readChan $ tickChannel state
-      sendMulti (outSocket state) serializeTick tick
+      sendMulti (outSocket state) $ fromList . map BL.toStrict $ serializeTick tick
 
 startQuoteSourceServer :: BoundedChan Tick -> Context -> String -> IO QuoteSourceServer
 startQuoteSourceServer chan c ep = do
@@ -33,8 +41,11 @@ startQuoteSourceServer chan c ep = do
     ctx = c,
     outSocket = sock,
     tickChannel = chan,
-    serverThread = tid
+    serverThreadId = tid
   }
   stid <- forkIO $ serverThread state
-  return $ state { serverThread = stid }
+  return $ state { serverThreadId = stid }
+
+stopQuoteSourceServer :: QuoteSourceServer -> IO ()
+stopQuoteSourceServer server = killThread $ serverThreadId server