main.hs

{-# LANGUAGE OverloadedStrings #-}

module Main (main) where

import Control.Concurrent (forkIO)
import Control.Monad (unless)
import Data.ByteString.Char8 (pack)
import System.Environment (getArgs)
import System.Timeout (timeout)
import System.ZMQ3


-- Sends a series of messages "1", "2", ... via a Push socket
pusher :: IO ()
pusher = withContext 1 $ \c -> withSocket c Push $ \s -> do
  bind s "tcp://*:5555"
  mapM_ (\i -> send s [] $ pack $ show i) ([1..] :: [Int])

-- Receives messages via a Pull socket using `System.Timeout.timeout` with
-- `maxtime`. After a `receive` call times out, the next successful call is
-- followed by printing the messages received adjacent to the timeout(s).
-- Setting a low value of `maxtime` demonstrates the possibility of losing
-- messages.
puller :: Int -> IO ()
puller maxtime = withContext 1 $ \c -> withSocket c Pull $ \s -> do
  connect s "tcp://localhost:5555"
  go s "0" (0 :: Int)
  where
    go s lastmsg nmisses = do
      mmsg <- timeout maxtime $ receive s
      case mmsg of
        Just msg -> do
          unless (nmisses == 0) $ putStrLn $
            show nmisses ++ " misses: " ++ show lastmsg ++ " -> " ++ show msg
          go s msg 0
        Nothing -> go s lastmsg (nmisses + 1)

-- Use the first command line arg as `maxtime` for `puller`.
main :: IO ()
main = do
  _ <- forkIO pusher
  getArgs >>= puller . read . head

zmq-timeout.cabal

name: zmq-timeout
version: 0.0.0
build-type: Simple
cabal-version: >= 1.9.2

executable main
  main-is: main.hs
  ghc-options: -Wall -threaded
  build-depends: base, bytestring, zeromq3-haskell