diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -16,10 +16,9 @@ You need Haskell GHC compiler installed (version 7.8.3 at least). You also need OpenSSL 1.0.2, since the ALPN feature and some very recent cypher-suites -are needed by HTTP/2. In this source distribution, I have set them to live in the +are needed by HTTP/2. this source distribution will try to find them at directory `/opt/openssl-1.0.2`, but you should be able to -alter the options using `cabal configure`. This package uses Haskell's foreign function -interface to interface with OpenSSL. +alter the options using `cabal configure`. This package uses Haskell's FFI to interface with OpenSSL. Provided that you have all the dependencies, you should be able to just do: @@ -38,6 +37,14 @@ - Version 0.1: Having something that can run. No unit-testing, nothing fancy. +- Version 0.2: Absolutely minimal amount of unit tests. + Pending: -- Version 0.2: Have some unit tests. A minimal amount of them +- Version 0.3: Better examples of usage. + +- Version 0.4: By-the-book stream state management. In particular, ensure + that we are not allowing frames to come off-order from the + other peer. + +- Version 0.5: Guaranties about early resource release... diff --git a/hs-src/SecondTransfer.hs b/hs-src/SecondTransfer.hs --- a/hs-src/SecondTransfer.hs +++ b/hs-src/SecondTransfer.hs @@ -32,22 +32,27 @@ {-# LANGUAGE OverloadedStrings #-} import SecondTransfer( CoherentWorker + , Footers , DataAndConclusion , tlsServeWithALPN , http2Attendant ) +import SecondTransfer.Http2( + makeSessionsContext + , defaultSessionsConfig + ) import Data.Conduit -saysHello :: 'DataAndConclusion' +saysHello :: DataAndConclusion saysHello = do - yield "Hello world!\\ns" + yield "Hello world!" -- No footers return [] -helloWorldWorker :: 'CoherentWorker' +helloWorldWorker :: CoherentWorker helloWorldWorker request = return ( [ (":status", "200") @@ -58,10 +63,12 @@ -- For this program to work, it should be run from the top of --- the developement directory, so that it has access to the toy --- certificates and keys defined there. +-- the developement directory. main = do - 'tlsServeWithALPN' + sessions_context <- makeSessionsContext defaultSessionsConfig + let + http2_attendant = http2Attendant sessions_context helloWorldWorker + tlsServeWithALPN "tests\/support\/servercert.pem" -- Server certificate "tests\/support\/privkey.pem" -- Certificate private key "127.0.0.1" -- On which interface to bind @@ -70,16 +77,13 @@ ("h2", http2_attendant) -- they may be slightly different, but for this -- test it doesn't matter. ] - 8000 - where - http2_attendant = http2Attendant helloWorldWorker + 8000 @ -`CoherentWorker` is the basic callback function that you need to implement. +`CoherentWorker` is the type of the basic callback function that you need to implement. The callback is used to handle all requests to the server on a given negotiated ALPN protocol. If you need routing functionality (and you most certainly will need it), you need -to build that functionality yourself or use one of the many Haskell libraries to that -end. +to build that functionality inside the callback. The above program uses a test certificate by a fake certificate authority. The certificate is valid for the server name ("authority", in HTTP\/2 lingo) www.httpdos.com. So, in order @@ -118,6 +122,9 @@ , FinalizationHeaders -- * Basic utilities for HTTP/2 servers + -- ** Configuration + + -- ** Callback types ,Attendant ,PullAction ,PushAction diff --git a/hs-src/SecondTransfer/Exception.hs b/hs-src/SecondTransfer/Exception.hs new file mode 100644 --- /dev/null +++ b/hs-src/SecondTransfer/Exception.hs @@ -0,0 +1,60 @@ +{-# LANGUAGE DeriveDataTypeable, ExistentialQuantification #-} +module SecondTransfer.Exception ( + -- * Exceptions thrown by the HTTP/2 sessions + HTTP2SessionException (..) + ,FramerException (..) + ,BadPrefaceException (..) + ) where + +import Control.Exception +import Data.Typeable + + +-- | Abstract exception. All HTTP/2 exceptions derive from here +data HTTP2SessionException = forall e . Exception e => HTTP2SessionException e + deriving Typeable + +instance Show HTTP2SessionException where + show (HTTP2SessionException e) = show e + +instance Exception HTTP2SessionException + +convertHTTP2SessionExceptionToException :: Exception e => e -> SomeException +convertHTTP2SessionExceptionToException = toException . HTTP2SessionException + +getHTTP2SessionExceptionFromException :: Exception e => SomeException -> Maybe e +getHTTP2SessionExceptionFromException x = do + HTTP2SessionException a <- fromException x + cast a + + + +-- | Abstract exception. Thrown when encoding/decoding of a frame fails +data FramerException = forall e . Exception e => FramerException e + deriving Typeable + +instance Show FramerException where + show (FramerException e) = show e + +instance Exception FramerException where + toException = convertHTTP2SessionExceptionToException + fromException = getHTTP2SessionExceptionFromException + +convertFramerExceptionToException :: Exception e => e -> SomeException +convertFramerExceptionToException = toException . FramerException + +getFramerExceptionFromException :: Exception e => SomeException -> Maybe e +getFramerExceptionFromException x = do + FramerException a <- fromException x + cast a + + + +-- | Thrown when the HTTP/2 connection prefix doesn't +-- match the expected prefix. +data BadPrefaceException = BadPrefaceException + deriving (Typeable, Show) + +instance Exception BadPrefaceException where + toException = convertFramerExceptionToException + fromException = getFramerExceptionFromException diff --git a/hs-src/SecondTransfer/Http2.hs b/hs-src/SecondTransfer/Http2.hs --- a/hs-src/SecondTransfer/Http2.hs +++ b/hs-src/SecondTransfer/Http2.hs @@ -1,5 +1,31 @@ module SecondTransfer.Http2( - http2Attendant + BadPrefaceException + ,http2Attendant + ,makeSessionsContext + ,defaultSessionsConfig + -- | Configuration information + ,SessionsConfig + -- | Context. You need to use `makeSessionsContext` to instance + -- one of this. + ,SessionsContext + ,SessionsCallbacks + ,ErrorCallback + ,sessionsCallbacks + + ,reportErrorCallback + -- | Lens to access the error callback. ) where import SecondTransfer.Http2.MakeAttendant(http2Attendant) +import SecondTransfer.Http2.Framer(BadPrefaceException) +import SecondTransfer.Http2.Session( + defaultSessionsConfig, + makeSessionsContext, + + SessionsConfig, + SessionsContext, + SessionsCallbacks, + ErrorCallback, + sessionsCallbacks, + reportErrorCallback + ) diff --git a/hs-src/SecondTransfer/Http2/Framer.hs b/hs-src/SecondTransfer/Http2/Framer.hs --- a/hs-src/SecondTransfer/Http2/Framer.hs +++ b/hs-src/SecondTransfer/Http2/Framer.hs @@ -4,7 +4,10 @@ DeriveDataTypeable, TemplateHaskell #-} {-# OPTIONS_HADDOCK hide #-} module SecondTransfer.Http2.Framer ( + BadPrefaceException, + wrapSession, + http2FrameLength, -- Not needed anywhere, but supress the warning about unneeded symbol closeAction @@ -14,43 +17,37 @@ import Control.Concurrent import Control.Exception -import qualified Control.Exception as E -import Control.Monad.IO.Class (liftIO) -import Control.Monad.Trans.Class (lift) -import Control.Monad.Trans.Reader -import qualified Control.Lens as L -import Control.Lens (view) -import Data.Binary (decode) -import qualified Data.ByteString as B -import qualified Data.ByteString.Lazy as LB +import qualified Control.Exception as E +import Control.Lens (view) +import qualified Control.Lens as L +import Control.Monad.IO.Class (liftIO) +import qualified Control.Monad.Catch as C +import Control.Monad.Trans.Class (lift) +import Control.Monad.Trans.Reader +import Data.Binary (decode) +import qualified Data.ByteString as B +import qualified Data.ByteString.Lazy as LB import Data.Conduit -import Data.Typeable (Typeable) -import Data.Foldable (find) - -import qualified Network.HTTP2 as NH2 +import Data.Foldable (find) +import qualified Network.HTTP2 as NH2 +-- Logging utilities +import System.Log.Logger -import qualified Data.HashTable.IO as H +import qualified Data.HashTable.IO as H -import SecondTransfer.Http2.Session +import SecondTransfer.Http2.Session import SecondTransfer.MainLoop.CoherentWorker (CoherentWorker) import qualified SecondTransfer.MainLoop.Framer as F -import SecondTransfer.MainLoop.PushPullType (Attendant, PullAction, - PushAction, CloseAction) +import SecondTransfer.MainLoop.PushPullType (Attendant, CloseAction, + PullAction, PushAction, IOProblem) import SecondTransfer.Utils (Word24, word24ToInt) +import SecondTransfer.Exception http2PrefixLength :: Int http2PrefixLength = B.length NH2.connectionPreface - -data BadPrefixException = BadPrefixException - deriving (Show, Typeable) - -instance Exception BadPrefixException - - - -- Let's do flow control here here .... type HashTable k v = H.CuckooHashTable k v @@ -77,10 +74,23 @@ , _stream2outputBytes :: HashTable GlobalStreamId (Chan LB.ByteString) , _defaultStreamWindow :: MVar Int + -- Wait variable to output bytes to the channel , _canOutput :: MVar CanOutput + -- Flag that says if the session has been unwound... if such, + -- threads are adviced to exit as early as possible + , _outputIsForbidden :: MVar Bool , _noHeadersInChannel :: MVar NoHeadersInChannel , _pushAction :: PushAction , _closeAction :: CloseAction + + -- Global id of the session, used for e.g. error reporting. + , _sessionId :: Int + + -- Sessions context, used for thing like e.g. error reporting + , _sessionsContext :: SessionsContext + + -- For GoAway frames + , _lastStream :: MVar Int } @@ -90,41 +100,66 @@ type FramerSession = ReaderT FramerSessionData IO -wrapSession :: CoherentWorker -> Attendant -wrapSession coherent_worker push_action pull_action close_action = do +wrapSession :: CoherentWorker -> SessionsContext -> Attendant +wrapSession coherent_worker sessions_context push_action pull_action close_action = do - let session_start = SessionStartData {} + let + session_id_mvar = view nextSessionId sessions_context - (session_input, session_output) <- http2Session coherent_worker session_start + new_session_id <- modifyMVarMasked + session_id_mvar + (\ session_id -> return (session_id+1, session_id)) + (session_input, session_output) <- (http2Session + coherent_worker new_session_id sessions_context) + -- TODO : Add type annotations.... - s2f <- H.new - s2o <- H.new - default_stream_size_mvar <- newMVar 65536 - can_output <- newMVar CanOutput - no_headers_in_channel <- newMVar NoHeadersInChannel + s2f <- H.new + s2o <- H.new + default_stream_size_mvar <- newMVar 65536 + can_output <- newMVar CanOutput + no_headers_in_channel <- newMVar NoHeadersInChannel + last_stream_id <- newMVar 0 + output_is_forbidden <- newMVar False -- We need some shared state let framer_session_data = FramerSessionData { - _stream2flow = s2f - ,_stream2outputBytes = s2o + _stream2flow = s2f + ,_stream2outputBytes = s2o ,_defaultStreamWindow = default_stream_size_mvar ,_canOutput = can_output ,_noHeadersInChannel = no_headers_in_channel ,_pushAction = push_action ,_closeAction = close_action + ,_sessionId = new_session_id + ,_sessionsContext = sessions_context + ,_lastStream = last_stream_id + ,_outputIsForbidden = output_is_forbidden } - forkIO $ close_on_error $ runReaderT (inputGatherer pull_action session_input ) framer_session_data - forkIO $ close_on_error $ runReaderT (outputGatherer session_output ) framer_session_data - return () + let + -- TODO: Dodgy exception handling here... + close_on_error session_id session_context comp = E.finally ( + E.catch comp (exc_handler session_id session_context)) close_action - where - close_on_error comp = E.finally comp close_action + exc_handler :: Int -> SessionsContext -> FramerException -> IO () + exc_handler x y e = do + modifyMVar_ output_is_forbidden (\ _ -> return True) + sessionExceptionHandler Framer_SessionComponent x y e + forkIO + $ close_on_error new_session_id sessions_context + $ runReaderT (inputGatherer pull_action session_input ) framer_session_data + forkIO + $ close_on_error new_session_id sessions_context + $ runReaderT (outputGatherer session_output ) framer_session_data + + return () + + http2FrameLength :: F.LengthCallback http2FrameLength bs | (B.length bs) >= 3 = let word24 = decode input_as_lbs :: Word24 @@ -172,11 +207,14 @@ (prefix, remaining) <- liftIO $ F.readLength http2PrefixLength pull_action if prefix /= NH2.connectionPreface - then - liftIO $ throwIO BadPrefixException + then do + sendGoAwayFrame NH2.ProtocolError + liftIO $ do + -- We just the the GoAway frame, although this is awfully early + -- and probably wrong + throwIO BadPrefaceException else return () - let source::Source FramerSession B.ByteString source = transPipe liftIO $ F.readNextChunk http2FrameLength remaining pull_action @@ -198,57 +236,75 @@ case error_or_frame of - Left some_error -> do - liftIO $ putStrLn $ "Got an error: " ++ (show some_error) + Left _ -> do + -- Got an error from the decoder... meaning that a frame could + -- not be decoded.... in this case we send a cancel session command + -- to the session. + liftIO $ errorM "HTTP2.Framer" "CouldNotDecodeFrame" + -- Send frames like GoAway and such... + lift $ sendGoAwayFrame NH2.ProtocolError + -- Inform the session that it can tear down itself + liftIO $ sendCommandToSession session_input CancelSession_SIC + -- Any resources remaining here can be disposed + lift $ releaseFramer + -- And end this thread + Right right_frame -> do + case right_frame of - Right (NH2.Frame (NH2.FrameHeader _ _ stream_id) (NH2.WindowUpdateFrame credit) ) -> do - -- Bookkeep the increase on bytes on that stream - -- liftIO $ putStrLn $ "Extra capacity for stream " ++ (show stream_id) - lift $ addCapacity (NH2.fromStreamIdentifier stream_id) (fromIntegral credit) - return () + (NH2.Frame (NH2.FrameHeader _ _ stream_id) (NH2.WindowUpdateFrame credit) ) -> do + -- Bookkeep the increase on bytes on that stream + -- liftIO $ putStrLn $ "Extra capacity for stream " ++ (show stream_id) + lift $ addCapacity (NH2.fromStreamIdentifier stream_id) (fromIntegral credit) + return () - Right frame@(NH2.Frame _ (NH2.SettingsFrame settings_list) ) -> do - -- Increase all the stuff.... - case find (\(i,_) -> i == NH2.SettingsInitialWindowSize) settings_list of + frame@(NH2.Frame _ (NH2.SettingsFrame settings_list) ) -> do + -- Increase all the stuff.... + case find (\(i,_) -> i == NH2.SettingsInitialWindowSize) settings_list of - Just (_, new_default_stream_size) -> do - old_default_stream_size_mvar <- view defaultStreamWindow - old_default_stream_size <- liftIO $ takeMVar old_default_stream_size_mvar - let general_delta = new_default_stream_size - old_default_stream_size - stream_to_flow <- view stream2flow - -- Add capacity to everybody's windows - liftIO $ - H.mapM_ ( - \ (k,v) -> if k /=0 - then writeChan v (AddBytes_FCM general_delta) - else return () ) - stream_to_flow + Just (_, new_default_stream_size) -> do + old_default_stream_size_mvar <- view defaultStreamWindow + old_default_stream_size <- liftIO $ takeMVar old_default_stream_size_mvar + let general_delta = new_default_stream_size - old_default_stream_size + stream_to_flow <- view stream2flow + -- Add capacity to everybody's windows + liftIO $ + H.mapM_ ( + \ (k,v) -> if k /=0 + then writeChan v (AddBytes_FCM general_delta) + else return () ) + stream_to_flow - -- And set a new value - liftIO $ putMVar old_default_stream_size_mvar new_default_stream_size + -- And set a new value + liftIO $ putMVar old_default_stream_size_mvar new_default_stream_size - Nothing -> - return () + Nothing -> + -- This is a silenced internal error + return () - -- And send the frame down to the session - liftIO $ sendFrameToSession session_input frame + -- And send the frame down to the session, so that session specific settings + -- can be applied. + liftIO $ sendFrameToSession session_input frame - Right a_frame -> do - liftIO $ sendFrameToSession session_input a_frame + a_frame@(NH2.Frame (NH2.FrameHeader _ _ stream_id) _ ) -> do + -- Update the keep of last stream + lift $ updateLastStream $ NH2.fromStreamIdentifier stream_id - -- tail recursion: go again... - consume + -- Send frame to the session + liftIO $ sendFrameToSession session_input a_frame + -- tail recursion: go again... + consume Nothing -> -- We may as well exit this thread return () +-- All the output frames come this way first outputGatherer :: SessionOutput -> FramerSession () outputGatherer session_output = do @@ -272,11 +328,9 @@ case command_or_frame of - Left cmd -> do - -- TODO: This is just a quickie behavior I dropped - -- here, semantics need to be different probably. - liftIO $ putStrLn $ "Received a command... terminating " ++ (show cmd) - + Left CancelSession_SOC -> do + -- The session wants to cancel things + releaseFramer Right ( p1@(NH2.EncodeInfo _ stream_idii _), p2@(NH2.DataFrame _) ) -> do -- This frame is flow-controlled... I may be unable to send this frame in @@ -296,13 +350,11 @@ loopPart - Right (p1, p2@(NH2.HeadersFrame _ _) ) -> do handleHeadersOfStream p1 p2 loopPart - Right (p1, p2@(NH2.ContinuationFrame _) ) -> do handleHeadersOfStream p1 p2 @@ -318,21 +370,50 @@ loopPart +updateLastStream :: GlobalStreamId -> FramerSession () +updateLastStream stream_id = do + last_stream_id_mvar <- view lastStream + liftIO $ modifyMVar_ last_stream_id_mvar (\ x -> return $ max x stream_id) + + + startStreamOutputQueue :: Int -> FramerSession (Chan LB.ByteString, Chan FlowControlCommand) startStreamOutputQueue stream_id = do -- New thread for handling outputs of this stream is needed bytes_chan <- liftIO newChan command_chan <- liftIO newChan + s2o <- view stream2outputBytes + liftIO $ H.insert s2o stream_id bytes_chan + s2c <- view stream2flow - liftIO $ H.insert s2c stream_id command_chan + + + liftIO $ H.insert s2c stream_id command_chan + + -- initial_cap_mvar <- view defaultStreamWindow initial_cap <- liftIO $ readMVar initial_cap_mvar + close_action <- view closeAction + sessions_context <- view sessionsContext + session_id' <- view SecondTransfer.Http2.Framer.sessionId + output_is_forbidden_mvar <- view outputIsForbidden -- And don't forget the thread itself + let + close_on_error session_id session_context comp = E.finally ( + E.catch comp (exc_handler session_id session_context)) close_action + + exc_handler :: Int -> SessionsContext -> IOProblem -> IO () + exc_handler x y e = do + -- Let's also decree that other streams don't even try + modifyMVar_ output_is_forbidden_mvar ( \ _ -> return True) + sessionExceptionHandler Framer_SessionComponent x y e + + read_state <- ask - liftIO $ forkIO $ runReaderT + liftIO $ forkIO $ close_on_error session_id' sessions_context $ runReaderT (flowControlOutput stream_id initial_cap "" command_chan bytes_chan) read_state @@ -341,6 +422,9 @@ -- This works in the output side of the HTTP/2 framing session, and it acts as a -- semaphore ensuring that headers are output without any interleaved frames. +-- +-- There are more synchronization mechanisms in the session, this does not act +-- alone. handleHeadersOfStream :: NH2.EncodeInfo -> NH2.FramePayload -> FramerSession () handleHeadersOfStream p1@(NH2.EncodeInfo _ _ _) frame_payload | (frameIsHeaderOfStream frame_payload) && (not $ frameEndsHeaders p1 frame_payload) = do @@ -360,8 +444,8 @@ | frameEndsHeaders p1 frame_payload = do -- I can only get here for a continuation frame after something else that is a headers no_headers <- view noHeadersInChannel - result <- liftIO $ tryPutMVar no_headers NoHeadersInChannel - liftIO $ putStrLn $ "Could put MVAR (yes must be): " ++ (show result) + liftIO $ putMVar no_headers NoHeadersInChannel + return () frameIsHeaderOfStream :: NH2.FramePayload -> Bool @@ -377,6 +461,8 @@ frameEndsHeaders _ _ = False +-- Push a frame into the output channel... this waits for the +-- channel to be free to send. pushFrame :: NH2.EncodeInfo -> NH2.FramePayload -> FramerSession () pushFrame p1 p2 = do @@ -384,23 +470,32 @@ sendBytes bs +sendGoAwayFrame :: NH2.ErrorCodeId -> FramerSession () +sendGoAwayFrame error_code = do + last_stream_id_mvar <- view lastStream + last_stream_id <- liftIO $ readMVar last_stream_id_mvar + pushFrame (NH2.EncodeInfo NH2.defaultFlags (NH2.toStreamIdentifier 0) Nothing) + (NH2.GoAwayFrame (NH2.toStreamIdentifier last_stream_id) error_code "") + + sendBytes :: LB.ByteString -> FramerSession () sendBytes bs = do push_action <- view pushAction can_output <- view canOutput liftIO $ do - bs `seq` takeMVar can_output - push_action bs - putMVar can_output CanOutput + bs `seq` + (C.bracket + (takeMVar can_output) + (\ _ -> push_action bs) + (\ c -> putMVar can_output c) + ) --- A thread in charge of doing flow control transmission --- TODO: Do session flow control..... +-- A thread in charge of doing flow control transmission....This sends already +-- formatted frames (ByteStrings), not the frames themselves. And it doesn't +-- mess with the structure of the packets. flowControlOutput :: Int -> Int -> LB.ByteString -> (Chan FlowControlCommand) -> (Chan LB.ByteString) -> FramerSession () flowControlOutput stream_id capacity leftovers commands_chan bytes_chan = do - -- Get some bytes to send - - if leftovers == "" then do -- Get more data (possibly block waiting for it) @@ -411,20 +506,26 @@ let amount = fromIntegral $ ((LB.length leftovers) - 9) if amount <= capacity then do + -- Is -- I can send ... if no headers are in process.... no_headers <- view noHeadersInChannel - liftIO $ takeMVar no_headers - sendBytes leftovers - -- liftIO $ putStrLn $ "Sent flow-controlled data for " ++ (show stream_id) - -- liftIO $ putStrLn $ "Capacity left " ++ (show (capacity - amount)) - liftIO $ putMVar no_headers NoHeadersInChannel - -- and tail-invoke + C.bracket + (liftIO $ takeMVar no_headers) + (\ _ -> liftIO $ putMVar no_headers NoHeadersInChannel) + (\ _ -> sendBytes leftovers ) flowControlOutput stream_id (capacity - amount) "" commands_chan bytes_chan else do -- I can not send because flow-control is full, wait for a command instead - liftIO $ putStrLn $ "Warning: channel flow-saturated " ++ (show stream_id) + -- liftIO $ putStrLn $ "Warning: channel flow-saturated " ++ (show stream_id) command <- liftIO $ readChan commands_chan case command of AddBytes_FCM delta_cap -> do -- liftIO $ putStrLn $ "Flow control delta_cap stream " ++ (show stream_id) flowControlOutput stream_id (capacity + delta_cap) leftovers commands_chan bytes_chan + + +releaseFramer :: FramerSession () +releaseFramer = do + -- Release any resources pending... + + return () diff --git a/hs-src/SecondTransfer/Http2/MakeAttendant.hs b/hs-src/SecondTransfer/Http2/MakeAttendant.hs --- a/hs-src/SecondTransfer/Http2/MakeAttendant.hs +++ b/hs-src/SecondTransfer/Http2/MakeAttendant.hs @@ -5,6 +5,7 @@ import SecondTransfer.Http2.Framer (wrapSession) +import SecondTransfer.Http2.Session (SessionsContext) import SecondTransfer.MainLoop.CoherentWorker import SecondTransfer.MainLoop.PushPullType ( --CloseAction, @@ -21,8 +22,8 @@ -- -- Given a `CoherentWorker`, this function wraps it with flow control, multiplexing, -- and state maintenance needed to run an HTTP/2 session. -http2Attendant :: CoherentWorker -> Attendant -http2Attendant coherent_worker push_action pull_action close_action = do +http2Attendant :: SessionsContext -> CoherentWorker -> Attendant +http2Attendant sessions_context coherent_worker push_action pull_action close_action = do let - attendant = wrapSession coherent_worker + attendant = wrapSession coherent_worker sessions_context attendant push_action pull_action close_action diff --git a/hs-src/SecondTransfer/Http2/Session.hs b/hs-src/SecondTransfer/Http2/Session.hs --- a/hs-src/SecondTransfer/Http2/Session.hs +++ b/hs-src/SecondTransfer/Http2/Session.hs @@ -8,44 +8,66 @@ ,getFrameFromSession ,sendFrameToSession ,sendCommandToSession + ,defaultSessionsConfig + ,sessionId + ,reportErrorCallback + ,sessionsCallbacks + ,nextSessionId + ,makeSessionsContext + ,sessionsConfig + ,sessionExceptionHandler ,CoherentSession ,SessionInput(..) ,SessionInputCommand(..) ,SessionOutput(..) - ,SessionStartData(..) + ,SessionOutputCommand(..) + ,SessionsContext(..) + ,SessionCoordinates(..) + ,SessionComponent(..) + ,SessionsCallbacks + ,SessionsConfig + ,ErrorCallback + + -- Internal stuff + ,OutputFrame + ,InputFrame ) where -- System grade utilities -import Control.Monad (forever) -import Control.Concurrent (forkIO, ThreadId) +import Control.Concurrent (ThreadId, forkIO) import Control.Concurrent.Chan -import Control.Monad.IO.Class (liftIO) +import Control.Exception (SomeException, throwTo) +import qualified Control.Exception as E +import Control.Monad (forever) +import Control.Monad.IO.Class (liftIO) import Control.Monad.Trans.Reader -import Control.Exception (throwTo) -import Data.Conduit -import Data.Conduit.List (foldMapM) -import qualified Data.ByteString as B import Control.Concurrent.MVar -import qualified Data.IntSet as NS -import qualified Data.HashTable.IO as H +import qualified Data.ByteString as B +import qualified Data.ByteString.Builder as Bu +import qualified Data.ByteString.Lazy as Bl +import Data.Conduit +import Data.Conduit.List (foldMapM) +import qualified Data.HashTable.IO as H +import qualified Data.IntSet as NS +import Data.Monoid as Mo import Control.Lens -- No framing layer here... let's use Kazu's Yamamoto library -import qualified Network.HTTP2 as NH2 -import qualified Network.HPACK as HP +import qualified Network.HPACK as HP +import qualified Network.HTTP2 as NH2 -- Logging utilities import System.Log.Logger -- Imports from other parts of the program -import SecondTransfer.MainLoop.CoherentWorker +import SecondTransfer.MainLoop.CoherentWorker import SecondTransfer.MainLoop.Tokens -import SecondTransfer.Utils (unfoldChannelAndSource) - +import SecondTransfer.Utils (unfoldChannelAndSource) +import SecondTransfer.Exception -- Unfortunately the frame encoding API of Network.HTTP2 is a bit difficult to -- use :-( @@ -102,20 +124,11 @@ getFrameFromSession (SessionOutput chan) = readChan chan --- Here is how we make a session -type SessionMaker = SessionStartData -> IO Session - - - --- Here is how we make a session wrapping a CoherentWorker -type CoherentSession = CoherentWorker -> SessionMaker - - type HashTable k v = H.CuckooHashTable k v -- Blaze builder could be more proper here... -type Stream2HeaderBlockFragment = HashTable GlobalStreamId B.ByteString +type Stream2HeaderBlockFragment = HashTable GlobalStreamId Bu.Builder type WorkerMonad = ReaderT WorkerThreadEnvironment IO @@ -134,22 +147,113 @@ deriving Show --- TODO: Put here information needed for the session to work -data SessionStartData = SessionStartData { - +-- | Information used to identify a particular session. +newtype SessionCoordinates = SessionCoordinates Int + deriving Show + +instance Eq SessionCoordinates where + (SessionCoordinates a) == (SessionCoordinates b) = a == b + +-- | Get/set a numeric Id from a `SessionCoordinates`. For example, to +-- get the session id with this, import `Control.Lens.(^.)` and then do +-- +-- @ +-- session_id = session_coordinates ^. sessionId +-- @ +-- +sessionId :: Functor f => (Int -> f Int) -> SessionCoordinates -> f SessionCoordinates +sessionId f (SessionCoordinates session_id) = + fmap (\ s' -> (SessionCoordinates s')) (f session_id) + + + +-- | Components at an individual session. Used to report +-- where in the session an error was produced. This interface is likely +-- to change in the future, as we add more metadata to exceptions +data SessionComponent = + SessionInputThread_SessionComponent + |SessionHeadersOutputThread_SessionComponent + |SessionDataOutputThread_SessionComponent + |Framer_SessionComponent + deriving Show + + +-- | Used by this session engine to report an error at some component, in a particular +-- session. +type ErrorCallback = (SessionComponent, SessionCoordinates, SomeException) -> IO () + +-- | Callbacks that you can provide your sessions to notify you +-- of interesting things happening in the server. +data SessionsCallbacks = SessionsCallbacks { + _reportErrorCallback :: Maybe ErrorCallback +} + +makeLenses ''SessionsCallbacks + + +-- | Configuration information you can provide to the session maker. +data SessionsConfig = SessionsConfig { + _sessionsCallbacks :: SessionsCallbacks +} + +-- makeLenses ''SessionsConfig + +-- | Lens to access sessionsCallbacks in the `SessionsConfig` object. +sessionsCallbacks :: Lens' SessionsConfig SessionsCallbacks +sessionsCallbacks f ( + SessionsConfig { + _sessionsCallbacks= s + }) = fmap (\ s' -> SessionsConfig {_sessionsCallbacks = s'}) (f s) + + +-- | Contains information that applies to all +-- sessions created in the program. Use the lenses +-- interface to access members of this struct. +-- +data SessionsContext = SessionsContext { + _sessionsConfig :: SessionsConfig + ,_nextSessionId :: MVar Int } -makeLenses ''SessionStartData +makeLenses ''SessionsContext +-- Here is how we make a session +type SessionMaker = SessionsContext -> IO Session + +-- Here is how we make a session wrapping a CoherentWorker +type CoherentSession = CoherentWorker -> SessionMaker + + +-- | Creates a default sessions context. Modify as needed using +-- the lenses interfaces +defaultSessionsConfig :: SessionsConfig +defaultSessionsConfig = SessionsConfig { + _sessionsCallbacks = SessionsCallbacks { + _reportErrorCallback = Nothing + } + } + + +-- Adds runtime data to a context, and let it work.... +makeSessionsContext :: SessionsConfig -> IO SessionsContext +makeSessionsContext sessions_config = do + next_session_id_mvar <- newMVar 1 + return $ SessionsContext { + _sessionsConfig = sessions_config, + _nextSessionId = next_session_id_mvar + } + data PostInputMechanism = PostInputMechanism (Chan (Maybe B.ByteString), InputDataStream) -- NH2.Frame != Frame data SessionData = SessionData { - _sessionInput :: Chan (Either SessionInputCommand InputFrame) + _sessionsContext :: SessionsContext + ,_sessionInput :: Chan (Either SessionInputCommand InputFrame) + -- We need to lock this channel occassionally so that we can order multiple -- header frames properly.... ,_sessionOutput :: MVar (Chan (Either SessionOutputCommand OutputFrame)) @@ -177,8 +281,10 @@ -- Worker thread register. This is a dictionary from stream id to -- the ThreadId of the thread with the worker thread. I use this to -- raise asynchronous exceptions in the worker thread if the stream - -- is cancelled by the client + -- is cancelled by the client. This way we get early finalization. ,_stream2WorkerThread :: HashTable Int ThreadId + + ,_sessionIdAtSession :: Int } @@ -186,8 +292,8 @@ -- v- {headers table size comes here!!} -http2Session :: CoherentWorker -> SessionStartData -> IO Session -http2Session coherent_worker _ = do +http2Session :: CoherentWorker -> Int -> SessionsContext -> IO Session +http2Session coherent_worker session_id sessions_context = do session_input <- newChan session_output <- newChan session_output_mvar <- newMVar session_output @@ -222,7 +328,8 @@ } let session_data = SessionData { - _sessionInput = session_input + _sessionsContext = sessions_context + ,_sessionInput = session_input ,_sessionOutput = session_output_mvar ,_toDecodeHeaders = decode_headers_table_mvar ,_toEncodeHeaders = encode_headers_table_mvar @@ -232,21 +339,30 @@ ,_streamsCancelled = cancelled_streams_mvar ,_stream2PostInputMechanism = stream2postinputmechanism ,_stream2WorkerThread = stream2workerthread + ,_sessionIdAtSession = session_id } + let + exc_handler :: SessionComponent -> HTTP2SessionException -> IO () + exc_handler component e = sessionExceptionHandler component session_id sessions_context e + exc_guard :: SessionComponent -> IO () -> IO () + exc_guard component action = E.catch action $ exc_handler component + -- Create an input thread that decodes frames... - forkIO $ runReaderT sessionInputThread session_data + forkIO $ exc_guard SessionInputThread_SessionComponent + $ runReaderT sessionInputThread session_data -- Create a thread that captures headers and sends them down the tube - forkIO $ runReaderT (headersOutputThread headers_output session_output_mvar) session_data + forkIO $ exc_guard SessionHeadersOutputThread_SessionComponent + $ runReaderT (headersOutputThread headers_output session_output_mvar) session_data -- Create a thread that captures data and sends it down the tube - forkIO $ dataOutputThread data_output session_output_mvar + forkIO $ exc_guard SessionDataOutputThread_SessionComponent + $ dataOutputThread data_output session_output_mvar - -- The two previous thread fill the session_output argument below (they write to it) + -- The two previous threads fill the session_output argument below (they write to it) -- the session machinery in the other end is in charge of sending that data through the -- socket. - return ( (SessionInput session_input), (SessionOutput session_output) ) @@ -278,7 +394,7 @@ Left CancelSession_SIC -> do -- Good place to tear down worker threads... Let the rest of the finalization - -- to somebody else.... + -- to the framer liftIO $ do H.mapM_ (\ (_, thread_id) -> do @@ -314,7 +430,6 @@ let source = postDataSourceFromMechanism mechanism return $ Just source else do - -- liftIO $ putStrLn "Headers end reqeust" return Nothing @@ -344,7 +459,8 @@ maybe_thread_id <- H.lookup stream2workerthread stream_id case maybe_thread_id of Nothing -> - errorM "HTTP2.Session" $ "Attention: could not find stream " ++ (show stream_id) ++ ("in threads register") + -- This is actually more like an internal error + error "InterruptingUnexistentStream" Just thread_id -> do throwTo thread_id StreamCancelledException @@ -352,7 +468,6 @@ continue - Right frame@(NH2.Frame (NH2.FrameHeader _ _ nh2_stream_id) (NH2.DataFrame somebytes)) -> do -- So I got data to process -- TODO: Handle end of stream @@ -390,7 +505,6 @@ continue - Right (NH2.Frame (NH2.FrameHeader _ flags _) (NH2.PingFrame _)) | NH2.testAck flags-> do -- Deal with pings: this is an Ack, so do nothing continue @@ -427,6 +541,7 @@ Right somethingelse -> do + -- An undhandled case here.... liftIO $ errorM "HTTP2.Session" $ "Received problematic frame: " liftIO $ errorM "HTTP2.Session" $ ".. " ++ (show somethingelse) @@ -581,9 +696,9 @@ maybe_old_block <- liftIO $ H.lookup ht global_stream_id new_block <- return $ case maybe_old_block of - Nothing -> bytes + Nothing -> Bu.byteString bytes - Just something -> something `B.append` bytes + Just something -> something `mappend` (Bu.byteString bytes) liftIO $ H.insert ht global_stream_id new_block @@ -592,7 +707,7 @@ getHeaderBytes global_stream_id = do ht <- view stream2HeaderBlockFragment Just bytes <- liftIO $ H.lookup ht global_stream_id - return bytes + return $ Bl.toStrict $ Bu.toLazyByteString bytes frameIsHeaderOfStream :: InputFrame -> Maybe (GlobalStreamId, B.ByteString) @@ -762,3 +877,21 @@ -- Restore output capability, so that other pieces waiting can send... liftIO $ debugM "HTTP2.Session" $ "Output capability restored" liftIO $ putMVar session_output_mvar session_output + + +sessionExceptionHandler :: E.Exception e => SessionComponent -> Int -> SessionsContext -> e -> IO () +sessionExceptionHandler session_component session_id sessions_context e = do + let + getit = ( sessionsConfig . sessionsCallbacks . reportErrorCallback ) + maybe_error_callback = sessions_context ^. getit + error_tuple = ( + session_component, + SessionCoordinates session_id, + E.toException e + ) + case maybe_error_callback of + Nothing -> + errorM "HTTP2.Session" (show (e)) + + Just callback -> + callback error_tuple diff --git a/hs-src/SecondTransfer/MainLoop.hs b/hs-src/SecondTransfer/MainLoop.hs --- a/hs-src/SecondTransfer/MainLoop.hs +++ b/hs-src/SecondTransfer/MainLoop.hs @@ -1,5 +1,10 @@ module SecondTransfer.MainLoop ( - -- * Callback types + -- * Callbacks + -- + -- | These callback make possible to separate the parts in layers. + -- The `Attendant` is a function that can push and pull bytes to/from + -- a transport (for example, a socket), but it is not concerned on how + -- those bytes are pushed or pulled. Attendant ,PullAction ,PushAction diff --git a/hs-src/SecondTransfer/MainLoop/Framer.hs b/hs-src/SecondTransfer/MainLoop/Framer.hs --- a/hs-src/SecondTransfer/MainLoop/Framer.hs +++ b/hs-src/SecondTransfer/MainLoop/Framer.hs @@ -1,6 +1,7 @@ {-# OPTIONS_HADDOCK hide #-} module SecondTransfer.MainLoop.Framer( readNextChunk + ,readNextChunkAndContinue ,readLength ,Framer @@ -39,16 +40,16 @@ readNextChunk length_callback input_leftovers gen = do let maybe_length = length_callback input_leftovers - readUpTo lo the_length | (B.length lo) >= the_length = + readUpTo_ lo the_length | (B.length lo) >= the_length = return $ B.splitAt the_length lo - readUpTo lo the_length = do + readUpTo_ lo the_length = do frag <- lift gen - readUpTo (lo `mappend` frag) the_length + readUpTo_ (lo `mappend` frag) the_length case maybe_length of Just the_length -> do -- Just need to read the rest .... - (package_bytes, newnewleftovers) <- readUpTo input_leftovers the_length + (package_bytes, newnewleftovers) <- readUpTo_ input_leftovers the_length yield package_bytes readNextChunk length_callback newnewleftovers gen @@ -59,20 +60,50 @@ readNextChunk length_callback new_leftovers gen +-- +readNextChunkAndContinue :: Monad m => + LengthCallback -- ^ How to know if we can split somewhere + -> B.ByteString -- ^ Input left-overs + -> m B.ByteString -- ^ Generator action + -> m (B.ByteString, B.ByteString) -- ^ Packet bytes and left-overs. +readNextChunkAndContinue length_callback input_leftovers gen = do + let + maybe_length = length_callback input_leftovers + + case maybe_length of + + Just the_length -> do + -- Just need to read the rest .... + (package_bytes, newnewleftovers) <- readUpTo gen input_leftovers the_length + return (package_bytes, newnewleftovers) + + Nothing -> do + -- Read a bit more + new_fragment <- gen + let new_leftovers = input_leftovers `mappend` new_fragment + readNextChunkAndContinue length_callback new_leftovers gen + + +readUpTo :: Monad m => m B.ByteString -> B.ByteString -> Int -> m (B.ByteString, B.ByteString) +readUpTo _ lo the_length | (B.length lo) >= the_length = + return $ B.splitAt the_length lo +readUpTo gen lo the_length = do + frag <- gen + readUpTo gen (lo `mappend` frag) the_length + + -- Some protocols, e.g., http/2, have the client transmit a fixed-length -- prefix. This function reads both that prefix and returns whatever get's -- trapped up there.... readLength :: MonadIO m => Int -> m B.ByteString -> m (B.ByteString, B.ByteString) readLength the_length gen = - readUpTo mempty + readUpTo_ mempty where - readUpTo lo + readUpTo_ lo | (B.length lo) >= the_length = do -- liftIO $ putStrLn "Full read" return $ B.splitAt the_length lo | otherwise = do -- liftIO $ putStrLn $ "fragment read " ++ (show lo) frag <- gen - readUpTo (lo `mappend` frag) - - + readUpTo_ (lo `mappend` frag) diff --git a/hs-src/SecondTransfer/MainLoop/Internal.hs b/hs-src/SecondTransfer/MainLoop/Internal.hs new file mode 100644 --- /dev/null +++ b/hs-src/SecondTransfer/MainLoop/Internal.hs @@ -0,0 +1,11 @@ +module SecondTransfer.MainLoop.Internal( + readNextChunkAndContinue + ,http2FrameLength + ,OutputFrame + ,InputFrame + ) where + + +import SecondTransfer.MainLoop.Framer(readNextChunkAndContinue) +import SecondTransfer.Http2.Framer(http2FrameLength) +import SecondTransfer.Http2.Session(OutputFrame, InputFrame) diff --git a/hs-src/SecondTransfer/MainLoop/PushPullType.hs b/hs-src/SecondTransfer/MainLoop/PushPullType.hs --- a/hs-src/SecondTransfer/MainLoop/PushPullType.hs +++ b/hs-src/SecondTransfer/MainLoop/PushPullType.hs @@ -18,8 +18,10 @@ -- | Callback type to push data to a channel. Part of this -- interface is the abstract exception type IOProblem. Throw an --- instance of it to notify the session that the connection has --- been broken. +-- instance of it from here to notify the session that the connection has +-- been broken. There is no way to signal "normal termination", since +-- HTTP/2's normal termination can be observed at a higher level when a +-- GO_AWAY frame is seen. type PushAction = LB.ByteString -> IO () -- | Callback type to pull data from a channel. The same @@ -27,7 +29,7 @@ -- there. type PullAction = IO B.ByteString --- | Callback that the Session calls to realease resources +-- | Callback that the session calls to realease resources -- associated with the channels. Take into account that your -- callback should be able to deal with non-clean shutdowns -- also, for example, if the connection to the remote peer @@ -43,6 +45,10 @@ -- push, pull and close callbacks. In this library we supply callbacks -- for TLS sockets, so that you don't need to go through the drudgery -- of managing those yourself. +-- +-- Attendants encapsulate all the session book-keeping functionality, +-- which for HTTP/2 is quite complicated. You use the function `http2Attendant` +-- to create one of these from a `CoherentWorker`. type Attendant = PushAction -> PullAction -> CloseAction -> IO () @@ -58,8 +64,8 @@ instance Exception IOProblem -- | A concrete case of the above exception. Throw one of this --- if you don't want to implement your own type. Capture one --- `IOProblem` otherwise. +-- if you don't want to implement your own type. Use +-- `IOProblem` in catch signatures. data GenericIOProblem = GenericIOProblem deriving (Show, Typeable) diff --git a/second-transfer.cabal b/second-transfer.cabal --- a/second-transfer.cabal +++ b/second-transfer.cabal @@ -10,13 +10,13 @@ -- PVP summary: +-+------- breaking API changes -- | | +----- non-breaking API additions -- | | | +--- code changes with no API change -version : 0.1.0.0 +version : 0.2.0.0 synopsis : Second Transfer HTTP/2 web server description : Second Transfer HTTP/2 web server -homepage : www.zunzun.se +homepage : https://github.com/alcidesv/second-transfer license : BSD3 @@ -27,7 +27,6 @@ maintainer : alcidesv@zunzun.se copyright : Copyright 2015, Alcides Viamontes Esquivel - category : Network stability : experimental @@ -50,7 +49,7 @@ source-repository this type: git location: git@github.com:alcidesv/second-transfer.git - tag: 0.1.0.0 + tag: 0.2.0.0 library @@ -59,6 +58,8 @@ , SecondTransfer.MainLoop , SecondTransfer.MainLoop.OpenSSL_TLS , SecondTransfer.Http2 + , SecondTransfer.MainLoop.Internal + , SecondTransfer.Exception other-modules: SecondTransfer.MainLoop.CoherentWorker , SecondTransfer.MainLoop.PushPullType @@ -77,14 +78,14 @@ -- optparse-applicative -- aeson >= 0.8 -- connection >= 0.2 && < 0.3 - -- exceptions >= 0.6 + exceptions >= 0.8 && < 0.9, bytestring == 0.10.4.0, -- base64-bytestring >= 1.0 base16-bytestring >= 0.1.1, -- data-default == 0.5.3 -- tls >= 1.2 && < 1.3 -- data-default-class == 0.0.1 - network == 2.6.0.2, + network >= 2.6 && < 2.7, -- x509 == 1.5.0.1 -- certificate == 1.3.9 -- x509-store == 1.5.0 @@ -106,7 +107,7 @@ -- crypto-random == 0.0.8 -- cryptohash >= 0.11, -- streaming-commons >= 0.1 && < 0.2 - conduit >= 1.2.4, + conduit >= 1.2.4 && < 1.3, -- conduit-combinators >= 0.3.0 && < 0.4 transformers >=0.3 && <= 0.5, -- mmorph == 1.0.4 @@ -148,12 +149,31 @@ extra-lib-dirs: /opt/openssl-1.0.2/lib + + Test-Suite compiling-ok type : exitcode-stdio-1.0 main-is : compiling_ok.hs hs-source-dirs : tests/tests-hs-src/ default-language: Haskell2010 - build-depends : base >=4.7 && < 4.8, - second-transfer, - conduit >= 1.2.4 + build-depends : base >=4.7 && < 4.8 + , second-transfer + , conduit >= 1.2.4 ghc-options : -threaded + + + +Test-Suite hunit-tests + type : exitcode-stdio-1.0 + main-is : hunit_tests.hs + hs-source-dirs : tests/tests-hs-src + default-language: Haskell2010 + build-depends : base >=4.7 && < 4.8 + ,second-transfer + ,conduit >= 1.2.4 + ,lens >= 4.7 + ,HUnit >= 1.2 && < 1.3 + ,bytestring == 0.10.4.0 + ,http2 == 0.9.1 + ghc-options : -threaded + other-modules : SecondTransfer.Test.DecoySession diff --git a/tests/tests-hs-src/SecondTransfer/Test/DecoySession.hs b/tests/tests-hs-src/SecondTransfer/Test/DecoySession.hs new file mode 100644 --- /dev/null +++ b/tests/tests-hs-src/SecondTransfer/Test/DecoySession.hs @@ -0,0 +1,110 @@ +{-# LANGUAGE TemplateHaskell, OverloadedStrings #-} + +module SecondTransfer.Test.DecoySession where + + + +import Control.Concurrent +import Control.Concurrent.MVar +import Control.Concurrent.Chan +import Control.Exception +import Control.Lens (view, (^.)) +import qualified Control.Lens as L + +import qualified Network.HTTP2 as NH2 + +import qualified Data.ByteString as B +import qualified Data.ByteString.Lazy as LB + +import SecondTransfer.Http2 +import SecondTransfer.MainLoop +import SecondTransfer.MainLoop.Internal + + +data DecoySession = DecoySession { + _inputDataChannel :: Chan B.ByteString + ,_outputDataChannel :: Chan LB.ByteString + ,_sessionThread :: ThreadId + ,_remainingOutputBit:: MVar B.ByteString + ,_nh2Settings :: NH2.Settings + ,_sessionThrowed :: MVar Bool + } + + +L.makeLenses ''DecoySession + + +-- Supposed to be an HTTP/2 attendant +createDecoySession :: Attendant -> IO DecoySession +createDecoySession attendant = do + input_data_channel <- newChan + output_data_channel <- newChan + let + push_action :: PushAction + push_action = writeChan output_data_channel + pull_action :: PullAction + pull_action = readChan input_data_channel + close_action :: CloseAction + close_action = do + return () + + thread_id <- forkIO $ do + attendant push_action pull_action close_action + + remaining_output_bit <- newMVar "" + let session = DecoySession { + _inputDataChannel = input_data_channel, + _outputDataChannel = output_data_channel, + _sessionThread = thread_id, + _remainingOutputBit= remaining_output_bit, + _nh2Settings = NH2.defaultSettings + } + return session + + +-- Tell when we are done with a dession +sessionIsUndone :: DecoySession -> IO Bool +sessionIsUndone session = error "NotImplemented" + + +-- Send a frame to a session +sendFrameToSession :: DecoySession -> OutputFrame -> IO () +sendFrameToSession session (encode_info, frame_payload) = do + let + bs_list = NH2.encodeFrameChunks encode_info frame_payload + input_data_channel = session ^. inputDataChannel + mapM_ (\ x -> writeChan input_data_channel x ) bs_list + + +-- Read a frame from a session... if possible. It will block +-- until the frame comes out, but should fail if there is +-- an exception and/or the session is closed. +recvFrameFromSession :: DecoySession -> IO (Maybe NH2.Frame) +recvFrameFromSession decoy_session = do + let + output_data_channel = decoy_session ^. outputDataChannel + remaining_output_bit_mvar = decoy_session ^. remainingOutputBit + pull_action = fmap LB.toStrict $ readChan output_data_channel + settings = decoy_session ^. nh2Settings + remaining_output_bit <- takeMVar remaining_output_bit_mvar + (packet, rest) <- readNextChunkAndContinue http2FrameLength remaining_output_bit pull_action + putMVar remaining_output_bit_mvar rest + let + error_or_frame = NH2.decodeFrame settings packet + case error_or_frame of + Left _ -> return Nothing + Right frame -> return $ Just frame + + +-- Send raw data to a session +sendRawDataToSession :: DecoySession -> B.ByteString -> IO () +sendRawDataToSession decoy_session data_to_send = do + let + input_data_channel = decoy_session ^. inputDataChannel + writeChan input_data_channel data_to_send + + +-- Read raw data from a session. Normally blocks until data be available. +-- It returns Nothing when the session is to be considered finished. +recvRawDataFromSession :: DecoySession -> IO (Maybe B.ByteString) +recvRawDataFromSession decoy_session = error "NotImplemented" diff --git a/tests/tests-hs-src/compiling_ok.hs b/tests/tests-hs-src/compiling_ok.hs --- a/tests/tests-hs-src/compiling_ok.hs +++ b/tests/tests-hs-src/compiling_ok.hs @@ -6,6 +6,10 @@ , tlsServeWithALPN , http2Attendant ) +import SecondTransfer.Http2( + makeSessionsContext + , defaultSessionsConfig + ) import Data.Conduit @@ -30,6 +34,9 @@ -- For this program to work, it should be run from the top of -- the developement directory. main = do + sessions_context <- makeSessionsContext defaultSessionsConfig + let + http2_attendant = http2Attendant sessions_context helloWorldWorker tlsServeWithALPN "tests/support/servercert.pem" -- Server certificate "tests/support/privkey.pem" -- Certificate private key @@ -39,6 +46,4 @@ ("h2", http2_attendant) -- they may be slightly different, but for this -- test it doesn't matter. ] - 8000 - where - http2_attendant = http2Attendant helloWorldWorker + 8000 diff --git a/tests/tests-hs-src/hunit_tests.hs b/tests/tests-hs-src/hunit_tests.hs new file mode 100644 --- /dev/null +++ b/tests/tests-hs-src/hunit_tests.hs @@ -0,0 +1,16 @@ +import Test.HUnit + +import SecondTransfer.Test.DecoySession + +import Tests.HTTP2Session + + +tests = TestList [ + TestLabel "testPrefaceChecks" testPrefaceChecks, + TestLabel "testPrefaceChecks2" testPrefaceChecks2 + ] + + +main = do + runTestTT tests + return ()