Support for RPC across ZeroMQ

msgpack-idl/Language/MessagePack/IDL/CodeGen/Cpp.hs

@@ -83,8 +83,19 @@ genTypeDecl _ MPType { .. } =
 typedef #{genType tyType} #{tyName};
 |]
 
+genTypeDecl _ MPEnum { ..} =
+  [lt|
+enum #{enumName} {
+    #{genEnum enumMem}
+};|]
+
+
 genTypeDecl _ _ = ""
 
+genEnum :: [(Int, T.Text)] -> LT.Text
+genEnum entries = LT.intercalate ",\n    " $ map enumEntry entries
+                where enumEntry (val, name) = [lt|#{name} = #{show val}|]
+
 genMsg name flds isExc =
   let fields = map f flds
       fs = map (maybe undefined fldName) $ sortField flds

msgpack-rpc/Network/MessagePackRpc/Server.hs

@@ -27,6 +27,7 @@ module Network.MessagePackRpc.Server (
   -- * RPC method types
   RpcMethod,
   RpcMethodType(..),
+  Endpoint(..),
   -- * Create RPC method
   fun,
   -- * Start RPC server
@@ -39,14 +40,17 @@ import Control.DeepSeq
 import Control.Exception as E
 import Control.Monad
 import Control.Monad.Trans
+import qualified Data.ByteString as B
 import qualified Data.ByteString.Lazy as BL
 import qualified Data.Conduit as C
 import qualified Data.Conduit.Binary as CB
 import qualified Data.Conduit.Attoparsec as CA
+import qualified Data.Attoparsec as A
 import Data.Maybe
 import Data.MessagePack
 import Network
 import System.IO
+import System.ZMQ
 
 import Prelude hiding (catch)
 
@@ -71,19 +75,24 @@ fromObject' o =
 fun :: RpcMethodType f => f -> RpcMethod
 fun = toRpcMethod
 
+data Endpoint = TCP Int
+              | ZeroMQ [String]
+              deriving Show
+
 -- | Start RPC server with a set of RPC methods.
-serve :: Int -- ^ Port number
+serve :: Endpoint -- ^ listen on this endpoint 
          -> [(String, RpcMethod)] -- ^ list of (method name, RPC method)
          -> IO ()
-serve port methods = withSocketsDo $ do
+
+serve (TCP port) methods = withSocketsDo $ do
   sock <- listenOn (PortNumber $ fromIntegral port)
   forever $ do
     (h, host, hostport) <- accept sock
     forkIO $
       (processRequests h `finally` hClose h) `catches`
       [ Handler $ \e ->
          case e of
-           CA.ParseError ["demandInput"] _ -> return ()
+           CA.ParseError ["demandInput"] _ _ -> return ()
            _ -> hPutStrLn stderr $ host ++ ":" ++ show hostport ++ ": " ++ show e
       , Handler $ \e ->
          hPutStrLn stderr $ host ++ ":" ++ show hostport ++ ": " ++ show (e :: SomeException)]
@@ -116,3 +125,36 @@ serve port methods = withSocketsDo $ do
           fail $ "method '" ++ methodName ++ "' not found"
         Just method ->
           method args
+
+serve (ZeroMQ endpoints) methods =
+  withContext 1 $ \ctx ->
+    withSocket ctx Rep $ \s -> do
+      mapM_ (bind s) endpoints
+      forever $ do
+        req  <- receive s []
+        resp <- processRequest req
+        send s ((B.concat . BL.toChunks) resp) []
+  where
+    processRequest req = 
+      case A.parseOnly get req of
+        Left _                             -> fail "Parsing failed."
+        Right (rtype, msgid, method, args) -> do
+          resp <- try $ getResponse rtype method args
+          case resp of
+            Left err ->
+              return $ pack (1 :: Int, msgid :: Int, show (err :: SomeException), ())
+            Right ret ->
+              return $ pack (1 :: Int, msgid :: Int, (), ret)
+
+    getResponse rtype method args = do
+      when (rtype /= (0 :: Int)) $
+        fail "request type is not 0"
+      r <- callMethod (method :: String) (args :: [Object])
+      r `deepseq` return r
+
+    callMethod methodName args =
+      case lookup methodName methods of
+        Nothing ->
+          fail $ "method '" ++ methodName ++ "' not found"
+        Just method ->
+          method args

msgpack-rpc/msgpack-rpc.cabal

@@ -23,10 +23,12 @@ Library
                    , network      >= 2.2 && < 2.4
                    , random       == 1.0.*
                    , mtl          == 2.0.*
-                   , conduit      >= 0.2 && < 0.5
+                   , conduit      >= 0.5
+                   , attoparsec
                    , attoparsec-conduit
                    , deepseq      >= 1.1 && < 1.4
                    , msgpack      == 0.7.*
+                   , zeromq-haskell >= 0.8.4
 
   Ghc-options:       -Wall