neptune-backend 0.1.2 → 0.2.1
raw patch · 8 files changed
+126/−23 lines, 8 filesdep +retrydep +unixdep +websockets
Dependencies added: retry, unix, websockets, wuss
Files
- README.md +0/−1
- examples/Main.hs +1/−1
- lib/Neptune/AbortHandler.hs +75/−0
- lib/Neptune/Channel.hs +2/−1
- lib/Neptune/Client.hs +40/−18
- lib/Neptune/OAuth.hs +1/−1
- lib/Neptune/Session.hs +1/−0
- neptune-backend.cabal +6/−1
README.md view
@@ -13,7 +13,6 @@ ## TODOs * Support logging text, image value. * Create project-* Register callback for the abort action from web console. * Support system channels * Get git information
examples/Main.hs view
@@ -16,6 +16,6 @@ main = do withNept "jiasen/sandbox" $ \_ experiment -> do- forM_ [1..10::Int] $ \i -> do+ forM_ [1..200::Int] $ \i -> do nlog experiment "counter" (fromIntegral (i * i) :: Double) threadDelay 1000000
+ lib/Neptune/AbortHandler.hs view
@@ -0,0 +1,75 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE QuasiQuotes #-}+module Neptune.AbortHandler where++import Control.Lens (_Right, (^?!))+import Control.Retry (constantDelay, recoverAll)+import Data.Aeson (FromJSON (..), Value (..),+ eitherDecode', (.:))+import Data.Aeson.Types (prependFailure, typeMismatch)+import Data.Text.Encoding (encodeUtf8)+import qualified Network.WebSockets as WS+import RIO+import RIO.Text as T (pack, unpack)+import Text.URI (RText, RTextLabel (..), mkURI)+import Text.URI.Lens+import qualified Wuss++import Neptune.Backend.Core (configHost)+import Neptune.Backend.Model (ExperimentId (..))+import Neptune.OAuth (oas_access_token)+import Neptune.Session (ClientToken (..), Experiment,+ NeptuneSession (..), exp_experiment_id)++abortListener :: NeptuneSession -> Experiment -> ThreadId -> IO ()+abortListener sess exp main_thread = do+ base_url <- mkURI (sess & _neptune_client_token & _ct_api_url)++ let host = base_url ^?! uriAuthority . _Right . authHost . unRText+ path = notif_path <> exp_id <> "/operations" :: String++ oauth_ref = sess & _neptune_oauth2++ run_listener = do+ oauth_current <- readMVar oauth_ref+ let oauth_token = oauth_current ^. oas_access_token+ Wuss.runSecureClientWith+ (T.unpack host) port path+ WS.defaultConnectionOptions+ [("Authorization", "Bearer " <> encodeUtf8 oauth_token)]+ (listener main_thread)++ recoverAll (constantDelay 500) $ \_ -> run_listener++ where+ port = 443+ notif_path = "/api/notifications/v1/experiments/"+ exp_id = T.unpack $ unExperimentId $ exp ^. exp_experiment_id+++data Message = MessageAbort { _msg_experiment_id :: Text }++instance FromJSON Message where+ parseJSON (Object v) = do+ typ <- v .: "messageType"+ case typ of+ "Abort" -> do+ body <- v .: "messageBody"+ MessageAbort <$> body .: "experimentId"+ _ -> fail $ "Unsupported message type" <> (T.unpack typ)+ parseJSON invalid =+ prependFailure "parsing Message failed, "+ (typeMismatch "Object" invalid)++data AbortException = AbortException+ deriving Show++instance Exception AbortException++listener :: ThreadId -> WS.ClientApp ()+listener main_thread conn = forever $ do+ msg <- WS.receiveData conn+ case eitherDecode' msg of+ Right (MessageAbort _) -> throwTo main_thread AbortException+ Left msg -> traceM (T.pack msg) >> return ()+
lib/Neptune/Channel.hs view
@@ -12,7 +12,8 @@ import Control.Concurrent.Event as E (isSet, set) import Control.Lens import Data.Typeable (Proxy (..), cast)-import RIO hiding (Lens', (^.))+import RIO hiding (Lens', (.~), (^.), (^..),+ (^?)) import qualified RIO.HashMap as M import qualified Neptune.Backend.API as NBAPI
lib/Neptune/Client.hs view
@@ -10,6 +10,8 @@ import Control.Concurrent (forkIO, killThread) import Control.Concurrent.Event as E (new, set, waitTimeout)+import Control.Exception (AsyncException (UserInterrupt),+ asyncExceptionFromException, try) import Control.Lens ((<&>), (^.)) import qualified Data.Text.Lazy as TL import qualified Data.Text.Lazy.Encoding as TL@@ -18,11 +20,14 @@ import Data.UUID.V4 as UUID (nextRandom) import qualified Network.HTTP.Client as NH import qualified Network.HTTP.Client.TLS as NH-import RIO hiding (Lens', (^.))+import RIO hiding (Lens', try, (^.)) import qualified RIO.HashMap as M import qualified RIO.Text as T import System.Envy (decodeEnv)+import System.Posix.Signals (Handler (Catch), installHandler,+ keyboardSignal) +import Neptune.AbortHandler (AbortException (..), abortListener) import qualified Neptune.Backend.API as NBAPI import Neptune.Backend.Client import Neptune.Backend.Core@@ -65,16 +70,22 @@ params "command" -- legacy (fromMaybe "Untitled" name)- tags){ experimentCreationParamsDescription = description }+ tags){ experimentCreationParamsDescription = description+ , experimentCreationParamsAbortable = Just True }+ let exp_id = ExperimentId (exp ^. experimentIdL) chan <- newTChanIO user_channels <- newTVarIO M.empty stop_flag <- E.new transmitter_flag <- E.new- let exp = Experiment exp_id chan user_channels stop_flag transmitter_flag undefined+ let exp = Experiment exp_id chan user_channels stop_flag transmitter_flag undefined undefined transmitter_thread <- forkIO $ transmitter session exp- return exp {_exp_transmitter = transmitter_thread} + parent_thread <- myThreadId+ abort_handler <- forkIO $ abortListener session exp parent_thread++ return exp {_exp_transmitter = transmitter_thread, _exp_abort_handler = abort_handler}+ where _mkParameter (ExperimentParamS name value) = do _id <- UUID.toText <$> UUID.nextRandom@@ -103,13 +114,25 @@ ses <- initNept project_qualified_name exp <- createExperiment ses Nothing Nothing [] [] [] + -- install an signal handler for CTRL-C, ensuring that an async-+ -- exception UserInterrupt is sent to the main thread+ main_thread <- myThreadId+ let interrupted = throwTo main_thread UserInterrupt+ installHandler keyboardSignal (Catch interrupted) Nothing+ result <- try (act ses exp) case result of Left (e :: SomeException) -> do- teardownNept ses exp ExperimentState'Failed (T.pack $ displayException e)+ let end_state = case fromException e of+ Just AbortException -> Nothing+ Nothing -> Just $+ case asyncExceptionFromException e of+ Just UserInterrupt -> (ExperimentState'Failed, "User interrupted.")+ Nothing -> (ExperimentState'Failed, T.pack $ displayException e)+ teardownNept ses exp end_state throwM e Right a -> do- teardownNept ses exp ExperimentState'Succeeded ""+ teardownNept ses exp (Just (ExperimentState'Succeeded, "")) return a -- | Initialize a neptune session@@ -120,7 +143,7 @@ ct@ClientToken{..} <- decodeEnv >>= either throwString return mgr <- NH.newManager NH.tlsManagerSettings- config0 <- withStderrLogging =<< newConfig+ config0 <- pure . withNoLogging =<< newConfig let api_endpoint = TL.encodeUtf8 (TL.fromStrict _ct_api_url) config = config0 { configHost = api_endpoint }@@ -154,10 +177,9 @@ -- | Teardown a neptune session teardownNept :: NeptuneSession -- ^ session -> Experiment -- ^ experiment- -> ExperimentState -- ^ completion state- -> Text -- ^ completion message+ -> Maybe (ExperimentState, Text) -- ^ completion state & message -> IO ()-teardownNept NeptuneSession{..} experiment state msg = do+teardownNept NeptuneSession{..} experiment state_msg = do E.set (experiment ^. exp_stop_flag) -- wait at most 5 seconds done <- E.waitTimeout (experiment ^. exp_transmitter_flag) 5000000@@ -166,11 +188,11 @@ killThread $ experiment ^. exp_transmitter killThread $ _neptune_oauth2_refresh - _ <- _neptune_dispatch $ NBAPI.markExperimentCompleted- (ContentType MimeJSON)- (Accept MimeNoContent)- (mkCompletedExperimentParams state msg)- (experiment ^. exp_experiment_id) :: IO NoContent-- return ()-+ case state_msg of+ Just (state, msg) ->+ void (_neptune_dispatch $ NBAPI.markExperimentCompleted+ (ContentType MimeJSON)+ (Accept MimeNoContent)+ (mkCompletedExperimentParams state msg)+ (experiment ^. exp_experiment_id) :: IO NoContent)+ Nothing -> return ()
lib/Neptune/OAuth.hs view
@@ -17,7 +17,7 @@ ReqBodyUrlEnc (..), defaultHttpConfig, jsonResponse, req, responseBody, runReq, useHttpsURI, (=:))-import RIO hiding (Lens', (^.))+import RIO hiding (Lens', (.~), (^.)) import qualified RIO.Text as T import Text.URI (mkURI) import qualified Web.JWT as JWT
lib/Neptune/Session.hs view
@@ -77,6 +77,7 @@ , _exp_stop_flag :: E.Event -- ^ Event flag to indicate the end of the session , _exp_transmitter_flag :: E.Event -- ^ Event flag to indicate completion of transmission , _exp_transmitter :: ThreadId -- ^ Background thread for transmission+ , _exp_abort_handler :: ThreadId } class (Typeable a, Show a) => NeptDataType a where
neptune-backend.cabal view
@@ -1,6 +1,6 @@ cabal-version: 2.2 name: neptune-backend-version: 0.1.2+version: 0.2.1 synopsis: Neptune Client description: . Client library for calling the Neptune API based on http-client.@@ -63,6 +63,10 @@ , envy , uuid , concurrent-extra+ , websockets+ , wuss+ , unix+ , retry other-modules: Paths_neptune_backend autogen-modules:@@ -74,6 +78,7 @@ Neptune.OAuth Neptune.Session Neptune.Utils+ Neptune.AbortHandler Neptune.Backend.API Neptune.Backend.API.ApiDefault Neptune.Backend.Client