diff --git a/app/SimpleConsumer.hs b/app/SimpleConsumer.hs --- a/app/SimpleConsumer.hs +++ b/app/SimpleConsumer.hs @@ -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 diff --git a/hs-asapo.cabal b/hs-asapo.cabal --- a/hs-asapo.cabal +++ b/hs-asapo.cabal @@ -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 diff --git a/lib/Asapo/Either/Producer.hs b/lib/Asapo/Either/Producer.hs --- a/lib/Asapo/Either/Producer.hs +++ b/lib/Asapo/Either/Producer.hs @@ -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 diff --git a/lib/Asapo/Producer.hs b/lib/Asapo/Producer.hs --- a/lib/Asapo/Producer.hs +++ b/lib/Asapo/Producer.hs @@ -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 ::