hs-asapo 0.9.0 → 0.9.1

raw patch · 4 files changed

+87/−46 lines, 4 filesdep ~bytestringdep ~textPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependency ranges changed: bytestring, text

API changes (from Hackage documentation)

Files

• app/SimpleConsumer.hs +31/−14
• hs-asapo.cabal +11/−7
• lib/Asapo/Either/Producer.hs +44/−24
• lib/Asapo/Producer.hs +1/−1

@@ -1,21 +1,26 @@ {-# LANGUAGE BlockArguments #-} {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}  module Main (main) where  import Asapo.Consumer-import Control.Applicative (Applicative ((<*>)))+import Control.Applicative (Applicative ((<*>)), pure)+import Control.Exception (Exception, IOException, SomeException, catch) import Control.Monad (forM_, (=<<), (>>=)) import Data.Bool (Bool (True)) import Data.Either (Either (Left, Right))+import Data.Foldable (for_) import Data.Function (($)) import Data.Functor ((<$>))-import Data.Maybe (Maybe (Nothing), fromMaybe)+import Data.Int (Int)+import Data.Maybe (Maybe (Just, Nothing), fromMaybe) import Data.Semigroup (Semigroup ((<>))) import Data.Text (Text, pack) import Data.Text.Encoding (decodeUtf8) import qualified Data.Text.IO as TIO import Data.Time.Clock (secondsToNominalDiffTime)+import Data.Traversable (for) import qualified Options.Applicative as Opt import System.IO (IO) import Text.Show (Show (show))@@ -26,7 +31,8 @@  data Options = Options   { optionsServerName :: Text,-    optionsWithFilesystem :: Bool+    optionsWithFilesystem :: Bool,+    optionsMessageId :: Maybe Int   }  optionsParser :: Opt.Parser Options@@ -34,6 +40,7 @@   Options     <$> Opt.strOption (Opt.long "server-name")     <*> Opt.switch (Opt.long "with-filesystem")+    <*> Opt.option Opt.auto (Opt.long "message-id" <> Opt.value Nothing)  main :: IO () main = realMain =<< Opt.execParser opts@@ -47,7 +54,7 @@         )  realMain :: Options -> IO ()-realMain (Options serverName withFilesystem) = do+realMain (Options serverName withFilesystem messageId) = do   withConsumer     (ServerName serverName)     (SourcePath "")@@ -68,7 +75,9 @@       TIO.putStrLn "misc: setting timeout"       setTimeout consumer (secondsToNominalDiffTime 0.5) -      getBeamtimeMeta consumer >>= \meta -> TIO.putStrLn $ "beamtime metadata: " <> (fromMaybe "N/A" meta)+      TIO.putStrLn "getting beamtime metadata"+      beamtimeMeta <- getBeamtimeMeta consumer `catch` (\(e :: SomeException) -> pure Nothing)+      TIO.putStrLn $ "beamtime metadata: " <> fromMaybe "N/A" beamtimeMeta        TIO.putStrLn "listing all available streams:"       streams <- getStreamList consumer Nothing FilterAllStreams@@ -76,18 +85,26 @@         TIO.putStrLn $ "=> stream info " <> pack (show stream)         streamSize <- getCurrentSize consumer (streamInfoName stream)         TIO.putStrLn $ "   stream size: " <> pack (show streamSize)-        datasetCount <- getCurrentDatasetCount consumer (streamInfoName stream) IncludeIncomplete+        datasetCount <- getCurrentDatasetCount consumer (streamInfoName stream) IncludeIncomplete `catch` (\(e :: SomeException) -> pure 0)         TIO.putStrLn $ "   dataset count: " <> pack (show datasetCount) -      -- withGroupId consumer outputError \groupId -> do-      --   onSuccess "getNextMessageMeta" (getNextMessageMeta consumer (streamInfoName stream) groupId) \(messageMetaHandle, messageMeta) -> do-      --     TIO.putStrLn "   got message meta"-      --   onSuccess "getNextMessageMetaAndData" (getNextMessageMetaAndData consumer (streamInfoName stream) groupId) \(messageMetaHandle, messageMeta, messageData) -> do-      --     TIO.putStrLn "   got message"+        -- withGroupId consumer outputError \groupId -> do+        --   onSuccess "getNextMessageMeta" (getNextMessageMeta consumer (streamInfoName stream) groupId) \(messageMetaHandle, messageMeta) -> do+        --     TIO.putStrLn "   got message meta"+        --   onSuccess "getNextMessageMetaAndData" (getNextMessageMetaAndData consumer (streamInfoName stream) groupId) \(messageMetaHandle, messageMeta, messageData) -> do+        --     TIO.putStrLn "   got message" -      (meta, data') <- getMessageMetaAndDataById consumer (StreamName "default") (messageIdFromInt 156)-      TIO.putStrLn $ "meta: " <> pack (show meta)-      TIO.putStrLn $ "data: " <> decodeUtf8 data'+        case messageId of+          Just mid -> do+            (meta, data') <- getMessageMetaAndDataById consumer (streamInfoName stream) (messageIdFromInt mid)+            TIO.putStrLn $ "    meta: " <> pack (show meta)+            TIO.putStrLn $ "    data: " <> decodeUtf8 data'+          Nothing -> do+            TIO.putStrLn "    messages from stream:"+            withGroupId consumer \groupId -> do+              (meta, data') <- getNextMessageMetaAndData consumer (streamInfoName stream) groupId+              TIO.putStrLn $ "      meta: " <> pack (show meta)+              TIO.putStrLn $ "      data: " <> decodeUtf8 data'        TIO.putStrLn "misc: resending nacs"       resendNacs consumer True (secondsToNominalDiffTime 1) 10

@@ -1,6 +1,6 @@ cabal-version:  3.4 name:           hs-asapo-version:        0.9.0+version:        0.9.1 synopsis:       Haskell bindings for ASAP:O category:       System, FFI, Distributed Computing homepage:       https://github.com/pmiddend/hs-asapo@@ -46,7 +46,7 @@                  , Asapo.Consumer   build-depends:       base >=4.7 && <5-    , text >= 2.0.2 && < 2.1+    , text >= 2.0.2 && < 2.2     -- for timespec data type (is Storable)     , clock >= 0.8.4 && < 0.9     -- for higher-level UTCTime/LocalTime@@ -54,11 +54,12 @@     -- for RFC3339     , timerep >= 2.1.0 && < 2.2     -- for the HL interface to send-    , bytestring  >= 0.11.5 && < 0.12+    , bytestring  >= 0.11.5 && < 0.13    pkgconfig-depends: libasapo-consumer                    , libasapo-producer   hs-source-dirs:    lib+  default-language:  Haskell2010  -- Marked as "benchmark" so the dependencies on hackage are accurate benchmark simple-producer@@ -71,9 +72,10 @@   build-depends:       base >=4.7 && <5     , hs-asapo-    , text >= 2.0.2 && < 2.1+    , text >= 2.0.2 && < 2.2     , time >= 1.12.2 && < 1.13     , optparse-applicative >= 0.18.1 && < 0.19+  default-language:  Haskell2010      -- Marked as "benchmark" so the dependencies on hackage are accurate benchmark simple-consumer@@ -86,9 +88,10 @@   build-depends:       base >=4.7 && <5     , hs-asapo-    , text >= 2.0.2 && < 2.1+    , text >= 2.0.2 && < 2.2     , time >= 1.12.2 && < 1.13     , optparse-applicative+  default-language:  Haskell2010      -- copied from https://github.com/ulidtko/cabal-doctest test-suite doctests@@ -100,9 +103,10 @@     , clock >= 0.8.4 && < 0.9     , timerep >= 2.1.0 && < 2.2     , time >= 1.12.2 && < 1.13-    , bytestring  >= 0.11.5 && < 0.12-    , text >= 2.0.2 && < 2.1+    , bytestring  >= 0.11.5 && < 0.13+    , text >= 2.0.2 && < 2.2     , hs-asapo    ghc-options:      -Wall -threaded   hs-source-dirs:   tests+  default-language:  Haskell2010

@@ -1,4 +1,5 @@ {-# LANGUAGE BlockArguments #-}+{-# LANGUAGE ImportQualifiedPost #-} {-# LANGUAGE ScopedTypeVariables #-}  -- |@@ -119,8 +120,8 @@ import Control.Exception (bracket) import Data.Bits ((.|.)) import Data.Bool (Bool)-import qualified Data.ByteString as BS-import Data.ByteString.Unsafe (unsafeUseAsCString)+import Data.ByteString qualified as BS+import Data.ByteString.Internal qualified as BSI import Data.Either (Either (Left, Right)) import Data.Eq (Eq ((==))) import Data.Foldable (Foldable (elem))@@ -130,9 +131,10 @@ import Data.Ord ((>)) import Data.Text (Text) import Data.Time (NominalDiffTime)-import Data.Word (Word64)+import Data.Word (Word64, Word8) import Foreign (Storable (peek), alloca, castPtr) import Foreign.C.ConstPtr (ConstPtr (unConstPtr))+import Foreign.ForeignPtr (ForeignPtr, withForeignPtr) import Foreign.Ptr (Ptr) import System.IO (IO) import Text.Show (Show)@@ -472,8 +474,14 @@     responseError :: Maybe Error   } -sendRequestCallback :: (RequestResponse -> IO ()) -> Ptr () -> AsapoRequestCallbackPayloadHandle -> AsapoErrorHandle -> IO ()-sendRequestCallback simpleCallback _data payloadHandle errorHandle = do+sendRequestCallback ::+  (RequestResponse -> IO ()) ->+  Maybe (ForeignPtr Word8) ->+  Ptr () ->+  AsapoRequestCallbackPayloadHandle ->+  AsapoErrorHandle ->+  IO ()+sendRequestCallback simpleCallback _dataAsForeignPtr _data payloadHandle errorHandle = do   payloadText <- bracket (asapo_request_callback_payload_get_response payloadHandle) asapo_free_string_handle stringHandleToTextUnsafe   originalHeaderCPtr <- asapo_request_callback_payload_get_original_header payloadHandle   originalHeaderC <- peek (unConstPtr originalHeaderCPtr)@@ -481,8 +489,11 @@   errorHandle' <- checkErrorWithGivenHandle errorHandle ()   case errorHandle' of     Left e -> simpleCallback (RequestResponse payloadText originalHeader (Just e))-    _ -> simpleCallback (RequestResponse payloadText originalHeader Nothing)+    Right _ -> simpleCallback (RequestResponse payloadText originalHeader Nothing) +toJustForeignPtr :: (a, b, c) -> a+toJustForeignPtr (p, _offset, _length) = p+ -- | Send a message containing raw data. Due to newtype and enum usage, all parameter should be self-explanatory send ::   Producer ->@@ -507,21 +518,30 @@     datasetSize     autoIdFlag     (BS.length data')-    \messageHeaderHandle ->-      unsafeUseAsCString data' \data'' -> withConstText stream \streamC -> do-        requestCallback <- createRequestCallback (sendRequestCallback callback)-        ( fromIntegral-            <$>-          )-          <$> checkError-            ( asapo_producer_send-                producer-                messageHeaderHandle-                (castPtr data'')-                (convertSendFlags transferFlag storageFlag)-                streamC-                requestCallback+    \messageHeaderHandle -> do+      withConstText stream \streamC -> do+        let word8Ptr :: ForeignPtr Word8+            word8Ptr = toJustForeignPtr (BSI.toForeignPtr data')+        requestCallback <-+          createRequestCallback+            ( sendRequestCallback+                callback+                -- We pass this ptr to the callback to keep the GC from collecting it+                (Just word8Ptr)             )+        withForeignPtr word8Ptr \dataPtr ->+          ( fromIntegral+              <$>+          )+            <$> checkError+              ( asapo_producer_send+                  producer+                  messageHeaderHandle+                  (castPtr dataPtr)+                  (convertSendFlags transferFlag storageFlag)+                  streamC+                  requestCallback+              )  -- | Send a message containing a file. Due to newtype and enum usage, all parameter should be self-explanatory sendFile ::@@ -553,7 +573,7 @@     size     \messageHeaderHandle ->       withConstText fileNameToSend \fileNameToSendC -> withConstText stream \streamC -> do-        requestCallback <- createRequestCallback (sendRequestCallback callback)+        requestCallback <- createRequestCallback (sendRequestCallback callback Nothing)         ( fromIntegral             <$>           )@@ -570,7 +590,7 @@ -- | As the title says, send the "stream finished" flag sendStreamFinishedFlag :: Producer -> StreamName -> MessageId -> StreamName -> (RequestResponse -> IO ()) -> IO (Either Error Int) sendStreamFinishedFlag (Producer producer) (StreamName stream) (MessageId lastId) (StreamName nextStream) callback = do-  requestCallback <- createRequestCallback (sendRequestCallback callback)+  requestCallback <- createRequestCallback (sendRequestCallback callback Nothing)   withConstText stream \streamC -> withConstText nextStream \nextStreamC ->     (fromIntegral <$>)       <$> checkError@@ -589,7 +609,7 @@ -- | Send or extend beamtime metadata sendBeamtimeMetadata :: Producer -> Metadata -> MetadataIngestMode -> UpsertMode -> (RequestResponse -> IO ()) -> IO (Either Error Int) sendBeamtimeMetadata (Producer producer) (Metadata metadata) ingestMode upsertMode callback = do-  requestCallback <- createRequestCallback (sendRequestCallback callback)+  requestCallback <- createRequestCallback (sendRequestCallback callback Nothing)   (fromIntegral <$>)     <$> withConstText metadata \metadataC ->       checkError@@ -611,7 +631,7 @@ -- | Send or extend stream metadata sendStreamMetadata :: Producer -> Metadata -> MetadataIngestMode -> UpsertMode -> StreamName -> (RequestResponse -> IO ()) -> IO (Either Error Int) sendStreamMetadata (Producer producer) (Metadata metadata) ingestMode upsertMode (StreamName stream) callback = do-  requestCallback <- createRequestCallback (sendRequestCallback callback)+  requestCallback <- createRequestCallback (sendRequestCallback callback Nothing)   (fromIntegral <$>)     <$> withConstText metadata \metadataC -> withConstText stream \streamC ->       checkError

@@ -278,7 +278,7 @@   (RequestResponse -> IO ()) ->   IO Int send producer messageId fileName metadata datasetSubstream datasetSize autoIdFlag data' transferFlag storageFlag stream callback =-  maybeThrow (PlainProducer.send producer messageId fileName metadata datasetSubstream datasetSize autoIdFlag data' transferFlag storageFlag stream callback)+  maybeThrow (PlainProducer.send producer messageId fileName metadata datasetSubstream datasetSize autoIdFlag (data') transferFlag storageFlag stream callback)  -- | Send a message containing a file. Due to newtype and enum usage, all parameter should be self-explanatory sendFile ::