cloudi 1.8.0 → 2.0.0

raw patch · 4 files changed

+134/−53 lines, 4 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• LICENSE +1/−1
• cloudi.cabal +5/−4
• src/Foreign/CloudI.hs +122/−45
• src/Foreign/CloudI/Instance.hs +6/−3

@@ -1,6 +1,6 @@ MIT License
 
-Copyright (c) 2017-2019 Michael Truog <mjtruog at protonmail dot com>
+Copyright (c) 2017-2020 Michael Truog <mjtruog at protonmail dot com>
 
 Permission is hereby granted, free of charge, to any person obtaining a
 copy of this software and associated documentation files (the "Software"),

@@ -1,5 +1,5 @@ name:                cloudi-version:             1.8.0+version:             2.0.0 synopsis:            Haskell CloudI API description:         Haskell CloudI API (see https://cloudi.org for more details) homepage:            https://github.com/CloudI/cloudi_api_haskell@@ -7,7 +7,7 @@ license-file:        LICENSE author:              Michael Truog maintainer:          mjtruog at protonmail dot com-copyright:           2017-2019 Michael Truog+copyright:           2017-2020 Michael Truog category:            Foreign stability:           provisional build-type:          Simple@@ -15,9 +15,10 @@ tested-with:           GHC == 7.10.3                      , GHC == 8.0.2                      , GHC == 8.2.2-                     , GHC == 8.4.3+                     , GHC == 8.4.4                      , GHC == 8.6.5-                     , GHC == 8.8.1+                     , GHC == 8.8.3+                     , GHC == 8.10.1 extra-source-files:  README.markdown changelog.md  flag dev

@@ -5,7 +5,7 @@    MIT License -  Copyright (c) 2017-2019 Michael Truog <mjtruog at protonmail dot com>+  Copyright (c) 2017-2020 Michael Truog <mjtruog at protonmail dot com>    Permission is hereby granted, free of charge, to any person obtaining a   copy of this software and associated documentation files (the "Software"),@@ -72,9 +72,10 @@     , threadCreate     , threadsWait     , infoKeyValueParse+    , infoKeyValueNew     ) where -import Prelude hiding (init,length)+import Prelude hiding (init,length,(<>)) import Data.Bits (shiftL,(.|.)) import Data.Maybe (fromMaybe) import Data.Typeable (Typeable)@@ -97,6 +98,7 @@ import qualified Foreign.Erlang as Erlang import qualified Foreign.CloudI.Instance as Instance import qualified System.IO as SysIO+import qualified System.IO.Error as SysIOErr import qualified System.IO.Unsafe as Unsafe import qualified System.Posix.Env as POSIX (getEnv) type Array = IArray.Array@@ -158,6 +160,7 @@     | ReturnAsync (Instance.T s)     | ForwardSync (Instance.T s)     | ForwardAsync (Instance.T s)+    | Terminate (Instance.T s)     deriving (Show, Typeable)  instance Typeable s => Exception.Exception (Exception s)@@ -178,10 +181,14 @@  type Result a = Either String a +infixr 4 <>+(<>) :: Monoid.Monoid m => m -> m -> m+(<>) = Monoid.mappend+ -- | creates an instance of the CloudI API-api :: Typeable s => Int -> s ->+api :: Typeable s => Int -> s -> Maybe Bool ->     IO (Result (Instance.T s))-api threadIndex state = do+api threadIndex state terminateReturnValueOpt = do     SysIO.hSetEncoding SysIO.stdout SysIO.utf8     SysIO.hSetBuffering SysIO.stdout SysIO.LineBuffering     SysIO.hSetEncoding SysIO.stderr SysIO.utf8@@ -190,22 +197,25 @@     bufferSizeValue <- POSIX.getEnv "CLOUDI_API_INIT_BUFFER_SIZE"     case (protocolValue, bufferSizeValue) of         (Just protocol, Just bufferSizeStr) ->-            let bufferSize = read bufferSizeStr :: Int+            let terminateReturnValue = fromMaybe True terminateReturnValueOpt+                terminateException = not terminateReturnValue+                bufferSize = read bufferSizeStr :: Int                 fd = C.CInt $ fromIntegral (threadIndex + 3)                 useHeader = protocol /= "udp"-                timeoutTerminate' = 1000+                timeoutTerminate' = 10 -- TIMEOUT_TERMINATE_MIN                 initTerms = Erlang.OtpErlangAtom (Char8.pack "init")             in             case Erlang.termToBinary initTerms (-1) of                 Left err ->                     return $ Left $ show err                 Right initBinary -> do-                    api0 <- Instance.make state protocol fd-                        useHeader bufferSize timeoutTerminate'+                    api0 <- Instance.make state terminateException+                        protocol fd useHeader bufferSize timeoutTerminate'                     send api0 initBinary                     result <- pollRequest api0 (-1) False                     case result of                         Left err ->+                            -- Terminate exception not used here                             return $ Left err                         Right (_, api1) ->                             return $ Right api1@@ -240,7 +250,9 @@ -- | returns the number of subscriptions for a single service name pattern subscribeCount :: Typeable s => Instance.T s -> ByteString ->     IO (Result (Int, Instance.T s))-subscribeCount api0 pattern =+subscribeCount api0@Instance.T{+      Instance.terminateException = terminateException}+    pattern =     let subscribeCountTerms = Erlang.OtpErlangTuple             [ Erlang.OtpErlangAtom (Char8.pack "subscribe_count")             , Erlang.OtpErlangString pattern]@@ -253,7 +265,10 @@             result <- pollRequest api0 (-1) False             case result of                 Left err ->-                    return $ Left err+                    if err == terminateError && terminateException then+                        Exception.throwIO $ Terminate api0+                    else+                        return $ Left err                 Right (_, api1@Instance.T{Instance.subscribeCount = count}) ->                     return $ Right (count, api1) @@ -277,7 +292,8 @@     Maybe Int -> Maybe ByteString -> Maybe Int ->     IO (Result (ByteString, Instance.T s)) sendAsync api0@Instance.T{-      Instance.timeoutAsync = timeoutAsync'+      Instance.terminateException = terminateException+    , Instance.timeoutAsync = timeoutAsync'     , Instance.priorityDefault = priorityDefault}     name request timeoutOpt requestInfoOpt priorityOpt =     let timeout = fromMaybe timeoutAsync' timeoutOpt@@ -299,7 +315,10 @@             result <- pollRequest api0 (-1) False             case result of                 Left err ->-                    return $ Left err+                    if err == terminateError && terminateException then+                        Exception.throwIO $ Terminate api0+                    else+                        return $ Left err                 Right (_, api1@Instance.T{Instance.transId = transId}) ->                     return $ Right (transId, api1) @@ -308,7 +327,8 @@     Maybe Int -> Maybe ByteString -> Maybe Int ->     IO (Result (ByteString, ByteString, ByteString, Instance.T s)) sendSync api0@Instance.T{-      Instance.timeoutSync = timeoutSync'+      Instance.terminateException = terminateException+    , Instance.timeoutSync = timeoutSync'     , Instance.priorityDefault = priorityDefault}     name request timeoutOpt requestInfoOpt priorityOpt =     let timeout = fromMaybe timeoutSync' timeoutOpt@@ -330,7 +350,10 @@             result <- pollRequest api0 (-1) False             case result of                 Left err ->-                    return $ Left err+                    if err == terminateError && terminateException then+                        Exception.throwIO $ Terminate api0+                    else+                        return $ Left err                 Right (_, api1@Instance.T{                       Instance.responseInfo = responseInfo                     , Instance.response = response@@ -343,7 +366,8 @@     Maybe Int -> Maybe ByteString -> Maybe Int ->     IO (Result (Array Int ByteString, Instance.T s)) mcastAsync api0@Instance.T{-      Instance.timeoutAsync = timeoutAsync'+      Instance.terminateException = terminateException+    , Instance.timeoutAsync = timeoutAsync'     , Instance.priorityDefault = priorityDefault}     name request timeoutOpt requestInfoOpt priorityOpt =     let timeout = fromMaybe timeoutAsync' timeoutOpt@@ -365,7 +389,10 @@             result <- pollRequest api0 (-1) False             case result of                 Left err ->-                    return $ Left err+                    if err == terminateError && terminateException then+                        Exception.throwIO $ Terminate api0+                    else+                        return $ Left err                 Right (_, api1@Instance.T{Instance.transIds = transIds}) ->                     return $ Right (transIds, api1) @@ -521,7 +548,9 @@ recvAsync :: Typeable s => Instance.T s ->     Maybe Int -> Maybe ByteString -> Maybe Bool ->     IO (Result (ByteString, ByteString, ByteString, Instance.T s))-recvAsync api0@Instance.T{Instance.timeoutSync = timeoutSync'}+recvAsync api0@Instance.T{+      Instance.terminateException = terminateException+    , Instance.timeoutSync = timeoutSync'}     timeoutOpt transIdOpt consumeOpt =     let timeout = fromMaybe timeoutSync' timeoutOpt         transId = fromMaybe transIdNull transIdOpt@@ -540,7 +569,10 @@             result <- pollRequest api0 (-1) False             case result of                 Left err ->-                    return $ Left err+                    if err == terminateError && terminateException then+                        Exception.throwIO $ Terminate api0+                    else+                        return $ Left err                 Right (_, api1@Instance.T{                       Instance.responseInfo = responseInfo                     , Instance.response = response@@ -630,16 +662,18 @@                 requestInfo request timeout priority transId pid                 state api1             case callbackResultAsyncValue of+                Left (Terminate api2) ->+                    return $ ReturnI (empty, empty, state, setTerminate api2)+                Left (ReturnAsync api2) ->+                    return $ Finished api2                 Left (ReturnSync api2) -> do                     printException "Synchronous Call Return Invalid"-                    return $ Finished api2-                Left (ReturnAsync api2) ->+                    return $ Finished (setTerminate api2)+                Left (ForwardAsync api2) ->                     return $ Finished api2                 Left (ForwardSync api2) -> do                     printException "Synchronous Call Forward Invalid"-                    return $ Finished api2-                Left (ForwardAsync api2) ->-                    return $ Finished api2+                    return $ Finished (setTerminate api2)                 Right (Instance.ResponseInfo (v0, v1, v2, v3)) ->                     return $ ReturnI (v0, v1, v2, v3)                 Right (Instance.Response (v0, v1, v2)) ->@@ -659,16 +693,18 @@                 requestInfo request timeout priority transId pid                 state api1             case callbackResultSyncValue of+                Left (Terminate api2) ->+                    return $ ReturnI (empty, empty, state, setTerminate api2)                 Left (ReturnSync api2) ->                     return $ Finished api2                 Left (ReturnAsync api2) -> do                     printException "Asynchronous Call Return Invalid"-                    return $ Finished api2+                    return $ Finished (setTerminate api2)                 Left (ForwardSync api2) ->                     return $ Finished api2                 Left (ForwardAsync api2) -> do                     printException "Asynchronous Call Forward Invalid"-                    return $ Finished api2+                    return $ Finished (setTerminate api2)                 Right (Instance.ResponseInfo (v0, v1, v2, v3)) ->                     return $ ReturnI (v0, v1, v2, v3)                 Right (Instance.Response (v0, v1, v2)) ->@@ -685,7 +721,7 @@     callbackResultType <- case callbackResultValue of         Left exception -> do             printException $ show (exception :: Exception.SomeException)-            return $ Finished api1+            return $ ReturnI (empty, empty, state, api1)         Right callbackResult ->             return $ callbackResult     case requestType of@@ -720,11 +756,8 @@     cmd <- if cmd0 == 0 then Get.getWord32host else return cmd0     case () of       _ | cmd == messageTerm ->-            let api1 = api0{-                      Instance.terminate = True-                    , Instance.timeout = Just False} in             if external then-                return ([], api1)+                return ([], setTerminate api0)             else                 fail terminateError         | cmd == messageReinit -> do@@ -928,16 +961,15 @@ pollRequestLoop :: Typeable s =>     Instance.T s -> Int -> Bool -> Clock.NominalDiffTime ->     IO (Result (Bool, Instance.T s))-pollRequestLoop api0@Instance.T{-      Instance.socketHandle = socketHandle-    , Instance.bufferRecvSize = bufferRecvSize}-    timeout external pollTimer = do-    inputAvailable <- if bufferRecvSize > 0 then-            return True-        else-            SysIO.hWaitForInput socketHandle timeout-    if not inputAvailable then+pollRequestLoop api0 timeout external pollTimer = do+    inputAvailable <- pollWait api0 timeout+    if inputAvailable == Just False then         return $ Right (True, api0{Instance.timeout = Nothing})+    else if inputAvailable == Nothing then+        let api1 = api0{+              Instance.terminate = True+            , Instance.timeout = Nothing} in+        return $ Right (False, api1)     else do         (dataIn, _, api1) <- recv api0         dataResult <- pollRequestData api1 external dataIn@@ -973,7 +1005,10 @@       Instance.initializationComplete = initializationComplete     , Instance.terminate = terminate} timeout external =     if terminate then-        return $ Right (False, api0)+        if external then+            return $ Right (False, api0)+        else+            return $ Left terminateError     else do         pollTimer <- if timeout <= 0 then                 return 0@@ -992,6 +1027,23 @@         else             pollRequestLoopBegin api0 timeout external pollTimer +pollWait :: Typeable s => Instance.T s -> Int -> IO (Maybe Bool)+pollWait Instance.T{+      Instance.socketHandle = socketHandle+    , Instance.bufferRecvSize = bufferRecvSize}+    timeout = do+    if bufferRecvSize > 0 then+        return $ Just True+    else+        SysIOErr.catchIOError+            (do+             inputAvailable <- SysIO.hWaitForInput socketHandle timeout+             return $ Just inputAvailable)+            (\e -> if SysIOErr.isEOFError e then+                 return Nothing+             else+                 SysIOErr.ioError e)+ -- | blocks to process incoming CloudI service requests poll :: Typeable s => Instance.T s -> Int -> IO (Result (Bool, Instance.T s)) poll api0 timeout =@@ -1111,6 +1163,11 @@         else             return (t1, fromIntegral $ timeoutValue - elapsed) +setTerminate :: Instance.T s -> Instance.T s+setTerminate api0 =+    api0{ Instance.terminate = True+        , Instance.timeout = Just False}+ threadList :: Concurrent.MVar [Concurrent.MVar ()] threadList = Unsafe.unsafePerformIO (Concurrent.newMVar []) @@ -1147,8 +1204,8 @@             Concurrent.takeMVar done             threadsWait -textKeyValueParse :: ByteString -> Map ByteString [ByteString]-textKeyValueParse text =+textPairsParse :: ByteString -> Map ByteString [ByteString]+textPairsParse text =     let loop m [] = m         loop m [v] =             if v == ByteString.empty then@@ -1160,11 +1217,31 @@                 Nothing ->                     loop (Map.insert k [v] m) l'                 Just v' ->-                    loop (Map.insert k (v:v') m) l'+                    loop (Map.insert k (v' ++ [v]) m) l'     in     loop Map.empty (Char8.split '\0' text) --- | parses "text_pairs" in service request info+textPairsNew :: Map ByteString [ByteString] -> Maybe Bool -> ByteString+textPairsNew pairs responseOpt =+    let response = fromMaybe True responseOpt in+    if response && Map.size pairs == 0 then+        Char8.pack "\0"+    else+        let pair builder _ [] =+                builder+            pair builder k (v:l') =+                pair (builder <>+                    Builder.byteString k <> Builder.char8 '\0' <>+                    Builder.byteString v <> Builder.char8 '\0') k l'+        in+        LazyByteString.toStrict $ Builder.toLazyByteString $+            Map.foldlWithKey pair Monoid.mempty pairs++-- | decode service request info key/value data infoKeyValueParse :: ByteString -> Map ByteString [ByteString]-infoKeyValueParse = textKeyValueParse+infoKeyValueParse = textPairsParse++-- | encode service response info key/value data+infoKeyValueNew :: Map ByteString [ByteString] -> Maybe Bool -> ByteString+infoKeyValueNew = textPairsNew 

@@ -5,7 +5,7 @@    MIT License -  Copyright (c) 2017-2019 Michael Truog <mjtruog at protonmail dot com>+  Copyright (c) 2017-2020 Michael Truog <mjtruog at protonmail dot com>    Permission is hereby granted, free of charge, to any person obtaining a   copy of this software and associated documentation files (the "Software"),@@ -99,6 +99,7 @@ -- | an instance of the CloudI API data T s = T     { state :: !s+    , terminateException :: !Bool     , socketHandle :: !Handle     , useHeader :: !Bool     , initializationComplete :: !Bool@@ -144,11 +145,13 @@     socket <- makeSocket protocol fd     Socket.socketToHandle socket SysIO.ReadWriteMode -make :: s -> String -> C.CInt -> Bool -> Int -> Int -> IO (T s)-make state' protocol fd useHeader' bufferSize' timeoutTerminate' = do+make :: s -> Bool -> String -> C.CInt -> Bool -> Int -> Int -> IO (T s)+make state' terminateException'+    protocol fd useHeader' bufferSize' timeoutTerminate' = do     socketHandle' <- makeSocketHandle protocol fd     return $ T {           state = state'+        , terminateException = terminateException'         , socketHandle = socketHandle'         , useHeader = useHeader'         , initializationComplete = False