PipeReader: read from zmq socket

app/Main.hs

@@ -130,9 +130,10 @@ main = do
     pipeReaderThread ctx config =
       case (tickPipePath config, pipeReaderQsEndpoint config) of
         (Just pipe, Just qsep) -> do
+          infoM "main" $ "Pipe/QS: " ++ pipe ++ "/" ++ qsep
           tickChan <- newBoundedChan 10000
-          bracket (startPipeReader (T.pack pipe) tickChan) stopPipeReader (\_ -> do
+          bracket (startPipeReader ctx (T.pack pipe) tickChan) stopPipeReader (\_ -> do
-            bracket (startQuoteSourceServer tickChan ctx (T.pack qsep) (Just "global")) stopQuoteSourceServer (\_ -> threadDelay 1000000))
+            bracket (startQuoteSourceServer tickChan ctx (T.pack qsep) (Just "global")) stopQuoteSourceServer (\_ -> forever $ threadDelay 1000000))
         _ -> return ()
 
 

src/QuoteSource/PipeReader.hs

@@ -1,4 +1,4 @@
-{-# LANGUAGE MultiWayIf #-}
+{-# LANGUAGE MultiWayIf        #-}
 {-# LANGUAGE OverloadedStrings #-}
 
 module QuoteSource.PipeReader (
@@ -6,60 +6,64 @@ module QuoteSource.PipeReader (
   stopPipeReader
   ) where
 
-import Data.IORef
+import           ATrade.QuoteSource.Server
-import qualified Data.Text as T
+import           ATrade.Types
-import qualified Data.Map.Strict as M
+import           Control.Applicative
-import qualified Data.HashSet as HS
+import           Control.Concurrent             hiding (readChan, writeChan,
-import Data.Time.Clock
+                                                 writeList2Chan, yield)
-import Data.Time.Calendar
+import           Control.Concurrent.BoundedChan
-import ATrade.Types
+import           Control.Error.Util
-import Control.Monad
+import           Control.Exception
-import Control.Monad.Extra
+import           Control.Monad
-import Control.Monad.Loops
+import           Control.Monad.Extra
-import Data.Maybe
+import           Control.Monad.IO.Class
-import Foreign.Marshal.Alloc
+import           Control.Monad.Loops
-import qualified Data.Text.Foreign as FT
+import           Data.Attoparsec.Text
-import System.Win32.File
+import qualified Data.ByteString                as B
-import System.Win32.Types
+import           Data.Conduit                   hiding (connect)
-import Control.Concurrent.BoundedChan
+import           Data.Conduit.Attoparsec
-import Control.Concurrent hiding (readChan, writeChan, writeList2Chan, yield)
+import qualified Data.Conduit.List              as CL
-import Control.Exception
+import qualified Data.HashSet                   as HS
-import Control.Monad.IO.Class
+import           Data.IORef
-import Control.Applicative
+import qualified Data.Map.Strict                as M
-import Safe
+import           Data.Maybe
-import Control.Error.Util
+import qualified Data.Text                      as T
-import Data.Attoparsec.Text
+import           Data.Text.Encoding
-import Data.Conduit
+import qualified Data.Text.Foreign              as FT
-import qualified Data.Conduit.List as CL
+import           Data.Time.Calendar
-import Data.Conduit.Attoparsec
+import           Data.Time.Clock
-import ATrade.QuoteSource.Server
+import           Foreign.Marshal.Alloc
+import           Safe
+import           System.IO
+import           System.Log.Logger              (debugM, warningM)
+import           System.ZMQ4
+
 
 data PipeReaderHandle =
   PipeReaderHandle {
     prThreadId :: ThreadId,
-    running :: IORef Bool
+    running    :: IORef Bool
-                   } deriving (Eq) 
+                   } deriving (Eq)
 
-data DataLine = CurrentParamLine T.Text Double Integer Double Integer Integer Double Integer Integer 
+data DataLine = CurrentParamLine T.Text Double Integer Double Integer Integer Double Integer Integer
               | AllTradeLine T.Text Integer Double Integer UTCTime
+              deriving (Show, Eq)
 
 yieldJust :: Maybe a -> Source IO a
 yieldJust maybeV = do -- Probably already present in some library
   case maybeV of
-    Just v -> yield v
+    Just v  -> yield v
     Nothing -> return ()
 
-win32FileConduit :: HANDLE -> Source IO T.Text
+zmqSocketConduit :: (Receiver a) => Socket a -> Source IO T.Text
-win32FileConduit handle = do -- TODO actually i have to check if pipe is closed
+zmqSocketConduit sock = do
-  maybeStr <- liftIO $ allocaBytes 4096 $ \buf -> do
+  maybeStr <- liftIO $ do
-    r <- win32_ReadFile handle buf 4096 Nothing
+    bs <- receive sock
-    if r > 0
+    case decodeUtf8' bs of
-       then do
+      Left _  -> return Nothing
-         s <- FT.peekCStringLen (buf, fromEnum r)
+      Right v -> return (Just v)
-         return $ Just s
-       else return Nothing
   yieldJust maybeStr
-  win32FileConduit handle
+  zmqSocketConduit sock
 
 line2TickConduit :: Conduit DataLine IO Tick
 line2TickConduit = do
@@ -67,7 +71,7 @@ line2TickConduit = do
   ignoreCPSet <- liftIO $ newIORef HS.empty
   lastTimestamp <- liftIO $ newIORef $ UTCTime (fromGregorian 1970 1 1) 0
   awaitForever $ \line ->
-    case line of 
+    case line of
       CurrentParamLine tickerId last voltoday bid biddepth biddeptht offer offerdepth offerdeptht -> do
         ts <- liftIO $ readIORef lastTimestamp
         yieldTick tickerId BestBid ts (fromDouble bid) biddepth
@@ -90,11 +94,11 @@ line2TickConduit = do
       AllTradeLine tickerId flags price volume ts -> do
         liftIO $ writeIORef lastTimestamp ts
         if
-           | flags == 1 -> yieldTick tickerId LastTradePrice ts (fromDouble price) (-volume)
+           | flags == 1 -> yieldTick tickerId LastTradePrice ts (fromDouble price) volume
            | flags == 2 -> yieldTick tickerId LastTradePrice ts (fromDouble price) volume
            | otherwise  -> return ()
         liftIO $ atomicModifyIORef' ignoreCPSet (\s -> (HS.insert tickerId s, ()))
-        
+
   where
     yieldTick tickerId dtype ts val vol =
       yield $ Tick { security = tickerId,
@@ -102,20 +106,25 @@ line2TickConduit = do
                      timestamp = ts,
                      value = val,
                      volume = vol }
-       
+
 chanSink :: BoundedChan QuoteSourceServerData -> Sink Tick IO ()
-chanSink chan = awaitForever (\t -> liftIO $ writeChan chan (QSSTick t))
+chanSink chan = awaitForever
+  (\t -> liftIO $ do
+      writeChan chan (QSSTick t))
 
-startPipeReader :: T.Text -> BoundedChan QuoteSourceServerData -> IO PipeReaderHandle
+startPipeReader :: Context -> T.Text -> BoundedChan QuoteSourceServerData -> IO PipeReaderHandle
-startPipeReader pipeName tickChan = do
+startPipeReader ctx pipeEndpoint tickChan = do
-  f <- createFile (T.unpack pipeName) gENERIC_READ 0 Nothing oPEN_EXISTING 0 Nothing
+  debugM "PipeReader" $ "Trying to open pipe: " ++ T.unpack pipeEndpoint
-  when (f == iNVALID_HANDLE_VALUE) $ error $ "Unable to open pipe: " ++ T.unpack pipeName
+  s <- socket ctx Sub
+  connect s (T.unpack pipeEndpoint)
+  subscribe s B.empty
+  debugM "PipeReader" "Pipe opened"
   running' <- newIORef True
-  tid <- forkIO $ readerThread f running'
+  tid <- forkIO $ readerThread s running'
   return PipeReaderHandle { prThreadId = tid, running = running' }
     where
-      readerThread f running' = runConduit $ (win32FileConduit f) =$= conduitParser parseTrade =$= CL.map snd =$= line2TickConduit =$= chanSink tickChan
+      readerThread s running' = runConduit $ (zmqSocketConduit s) =$= conduitParserEither parseTrade =$= handleParseResult =$= line2TickConduit =$= chanSink tickChan
-      parseTrade = parseCurrentParam <|> parseAllTrade 
+      parseTrade = parseCurrentParam <|> parseAllTrade
       parseCurrentParam = do
         string "CT:"
         secName <- takeTill (== ':')
@@ -168,5 +177,12 @@ startPipeReader pipeName tickChan = do
         ms <- fromInteger <$> decimal
         return $ UTCTime (fromGregorian y mon day) $ h * 3600 + m * 60 + s + ms / 1000
 
+      handleParseResult = do
+        awaitForever $ \res ->
+          case res of
+            Left err -> liftIO $ warningM "PipeReader" $ "Can't parse: " ++ show err
+            Right (_, r) -> yield r
+
+
 stopPipeReader :: PipeReaderHandle -> IO ()
 stopPipeReader h = killThread (prThreadId h) >> writeIORef (running h) False