packages feed

second-transfer 0.5.2.2 → 0.5.3.1

raw patch · 19 files changed

+2178/−1815 lines, 19 files

Files

cbits/tlsinc.c view
@@ -456,7 +456,7 @@         // Now I set a few options....         /*SSL_CTX_set_verify(c->sslContext, NULL );*/         const long flags = -          SSL_OP_NO_SSLv2 | SSL_OP_NO_SSLv3 | SSL_OP_NO_TLSv1 | SSL_OP_NO_SESSION_RESUMPTION_ON_RENEGOTIATION | SSL_OP_NO_COMPRESSION;+          SSL_OP_NO_SSLv2 | SSL_OP_NO_SSLv3 | SSL_OP_NO_SESSION_RESUMPTION_ON_RENEGOTIATION | SSL_OP_NO_COMPRESSION;         // const long flags =          //     SSL_OP_NO_SESSION_RESUMPTION_ON_RENEGOTIATION | SSL_OP_NO_COMPRESSION;         SSL_CTX_set_options(c->sslContext, flags);@@ -469,9 +469,9 @@          // The only cipher supported by HTTP/2 ... sort of.         result = SSL_CTX_set_cipher_list(c->sslContext, -            "ECDHE-RSA-AES256-GCM-SHA384"  // <-- Good for Chrome and firefox-            ":ECDHE-RSA-AES128-GCM-SHA256" //  --  ibidem-            // ---- These suites will fail HTTP/2+            "ECDHE-RSA-AES128-GCM-SHA256" //  --  ibidem+            ":ECDHE-RSA-AES256-GCM-SHA384"  // <-- Good for Chrome and firefox+             // ---- These suites will fail HTTP/2             ":ECDHE-RSA-AES128-SHA256"     //  -- Not good for HTTP/2             ":ECDHE-RSA-AES-128-CBC-SHA256"             ":ECDHE-ECDSA-AES256-SHA384"
hs-src/SecondTransfer.hs view
@@ -69,13 +69,15 @@   helloWorldWorker :: CoherentWorker-helloWorldWorker request = return (-    [-        (":status", "200")-    ],-    [], -- No pushed streams-    saysHello-    )+helloWorldWorker (_request_headers, _maybe_post_data) = do +    dropIncomingData _maybe_post_data+    return (+        [+            (":status", "200")+        ],+        [], -- No pushed streams+        saysHello+        )   -- For this program to work, it should be run from the top of @@ -156,6 +158,7 @@     --   HTTP/2 server in a snap.     ,tlsServeWithALPN     ,tlsServeWithALPNAndFinishOnRequest+    ,dropIncomingData      ,TLSLayerGenericProblem(..)     ,FinishRequest(..)@@ -175,3 +178,4 @@ import           SecondTransfer.MainLoop import           SecondTransfer.MainLoop.CoherentWorker import           SecondTransfer.Types+import           SecondTransfer.Utils.DevNull           (dropIncomingData)
hs-src/SecondTransfer/Exception.hs view
@@ -15,6 +15,9 @@     ,IOProblem(..)     ,GenericIOProblem(..)     ,StreamCancelledException(..)++    -- * Internal exceptions+    ,HTTP2ProtocolException(..)     ) where   import           Control.Exception@@ -39,6 +42,14 @@     HTTP2SessionException a <- fromException x     cast a +-- | Concrete exception. Used internally to signal that the client violated+--   the protocol. Clients of the library shall never see this exception.+data HTTP2ProtocolException = HTTP2ProtocolException+    deriving (Typeable, Show)++instance Exception HTTP2ProtocolException where+    toException   = convertHTTP2SessionExceptionToException+    fromException = getHTTP2SessionExceptionFromException   -- | Abstract exception. Thrown when encoding/decoding of a frame fails
hs-src/SecondTransfer/Http1/Parse.hs view
@@ -32,9 +32,11 @@ import qualified Data.ByteString.Lazy                   as Lb import           Data.Char                              (toLower) import           Data.Maybe                             (isJust)+#ifndef IMPLICIT_MONOID import           Data.Monoid                            (mappend,                                                          mempty,                                                          mconcat)+#endif import qualified Data.Attoparsec.ByteString             as Ap  import           Data.Foldable                          (find)@@ -46,7 +48,7 @@ import           SecondTransfer.MainLoop.CoherentWorker (Headers) import           SecondTransfer.Utils                   (subByteString) -+import           Prelude   data IncrementalHttp1Parser = IncrementalHttp1Parser {
− hs-src/SecondTransfer/Http1/Session.cpphs 
@@ -1,108 +0,0 @@-{-# LANGUAGE OverloadedStrings #-}-{-# OPTIONS_HADDOCK hide #-}-module SecondTransfer.Http1.Session(-    http11Attendant-    ) where -- --- import qualified Control.Lens                           as L--- import           Control.Lens                           ( (^.) )-import           Control.Exception                     (catch)-import           Control.Concurrent                    (forkIO)--import qualified Data.ByteString                        as B--- import qualified Data.ByteString.Lazy                   as LB--- import           Data.ByteString.Char8                  (unpack)--- import qualified Data.ByteString.Builder                as Bu-import           Data.Conduit-import           Data.Conduit.List                      (consume)--- import           Data.Monoid                            (mconcat, mappend)--import           SecondTransfer.MainLoop.CoherentWorker (CoherentWorker,)-import           SecondTransfer.MainLoop.PushPullType   (Attendant)-import           SecondTransfer.Sessions.Internal       (SessionsContext, acquireNewSessionTag)---- Logging utilities-import           System.Log.Logger                      --import           SecondTransfer.Http1.Parse-import           SecondTransfer.Exception                (IOProblem)---- import           Debug.Trace                             (traceShow)----- | Session attendant that speaks HTTP/1.1--- -http11Attendant :: SessionsContext -> CoherentWorker -> Attendant-http11Attendant sessions_context coherent_worker -                push_action pull_action close_action -    = do -        new_session_tag <- acquireNewSessionTag sessions_context-        infoM "Session.Session_HTTP11" $ "Starting new session with tag: " ++(show new_session_tag)-        forkIO $ go new_session_tag (Just "")-        return ()-  where -    go :: Int -> Maybe B.ByteString -> IO ()-    go session_tag (Just leftovers) = do -        infoM "Session.Session_HTTP11" $ "(Re)Using session with tag: " ++(show session_tag)-        maybe_leftovers <- add_data newIncrementalHttp1Parser leftovers session_tag-        go session_tag maybe_leftovers--    go _ Nothing = -        return ()--    add_data :: IncrementalHttp1Parser  -> B.ByteString -> Int -> IO (Maybe B.ByteString)-    add_data parser bytes session_tag = do -        let -            completion = addBytes parser bytes -            -- completion = addBytes parser $ traceShow ("At session " ++ (show session_tag) ++ " Received: " ++ (unpack bytes) ) bytes-        case completion of --            MustContinue_H1PC new_parser -> do -                -- print "MustContinue_H1PC"-                catch -                    (do-                        new_bytes <- pull_action-                        r <- add_data new_parser new_bytes session_tag-                        return r-                    )-                    ( (\ _e -> do-                        -- This is a pretty harmless condition that happens -                        -- often when the remote peer closes the connection-                        debugM "Session.HTTP1" "Could not receive data"-                        close_action-                        return Nothing-                    ) :: IOProblem -> IO (Maybe B.ByteString) )--                --            OnlyHeaders_H1PC headers leftovers -> do -                -- print "OnlyHeaders_H1PC"-                -- Ready for action...-                -- ATTENTION: Not use for pushed streams here....-                -- We must decide what to do if the user return those-                -- anyway.-                (response_headers, _, data_and_conclusion) <- coherent_worker (headers, Nothing)-                (_, fragments) <- runConduit $ fuseBoth data_and_conclusion consume -                let -                    response_text =-                        serializeHTTPResponse response_headers fragments--                catch -                    (do-                        push_action response_text-                        return $ Just leftovers-                    )-                    ((\ _e -> do-                        debugM "Session.HTTP1" "Session abandoned"-                        close_action-                        return Nothing-                    ) :: IOProblem -> IO (Maybe B.ByteString) )--            HeadersAndBody_H1PC _headers _stopcondition _recv_leftovers -> do-                -- print "HeadersAndBody_H1PC"-                -- Let's see if I can go through the basic movements first, then through -                -- more complicated things.-                -- TODO: Implement posts and other requests with bodies....-                close_action-                error "NotImplemented requests with bodies"
+ hs-src/SecondTransfer/Http1/Session.hs view
@@ -0,0 +1,136 @@+{-# LANGUAGE OverloadedStrings #-}+{-# OPTIONS_HADDOCK hide #-}+module SecondTransfer.Http1.Session(+    http11Attendant+    ) where ++ +import           Control.Lens+import           Control.Exception                       (catch)+import           Control.Concurrent                      (forkIO)++import qualified Data.ByteString                         as B+-- import qualified Data.ByteString.Lazy                   as LB+-- import           Data.ByteString.Char8                  (unpack)+-- import qualified Data.ByteString.Builder                as Bu+import           Data.Conduit+import           Data.Conduit.List                       (consume)+-- import           Data.Monoid                            (mconcat, mappend)++import           SecondTransfer.MainLoop.CoherentWorker  (CoherentWorker,Headers)+import           SecondTransfer.MainLoop.PushPullType    (Attendant)+import           SecondTransfer.Sessions.Internal        (SessionsContext, acquireNewSessionTag)++-- Logging utilities+import           System.Log.Logger                      ++import           SecondTransfer.Http1.Parse+import           SecondTransfer.Exception                (IOProblem)+import           SecondTransfer.Sessions.Config+import           SecondTransfer.Sessions.Internal        (sessionsConfig)+import qualified SecondTransfer.Utils.HTTPHeaders        as He++-- import           Debug.Trace                             (traceShow)+++-- | Session attendant that speaks HTTP/1.1+-- +http11Attendant :: SessionsContext -> CoherentWorker -> Attendant+http11Attendant sessions_context coherent_worker +                push_action pull_action close_action +    = do +        new_session_tag <- acquireNewSessionTag sessions_context+        infoM "Session.Session_HTTP11" $ "Starting new session with tag: " ++(show new_session_tag)+        forkIO $ go new_session_tag (Just "")+        return ()+  where +    go :: Int -> Maybe B.ByteString -> IO ()+    go session_tag (Just leftovers) = do +        infoM "Session.Session_HTTP11" $ "(Re)Using session with tag: " ++(show session_tag)+        maybe_leftovers <- add_data newIncrementalHttp1Parser leftovers session_tag+        go session_tag maybe_leftovers++    go _ Nothing = +        return ()++    add_data :: IncrementalHttp1Parser  -> B.ByteString -> Int -> IO (Maybe B.ByteString)+    add_data parser bytes session_tag = do +        let +            completion = addBytes parser bytes +            -- completion = addBytes parser $ traceShow ("At session " ++ (show session_tag) ++ " Received: " ++ (unpack bytes) ) bytes+        case completion of ++            MustContinue_H1PC new_parser -> do +                -- print "MustContinue_H1PC"+                catch +                    (do+                        new_bytes <- pull_action+                        r <- add_data new_parser new_bytes session_tag+                        return r+                    )+                    ( (\ _e -> do+                        -- This is a pretty harmless condition that happens +                        -- often when the remote peer closes the connection+                        debugM "Session.HTTP1" "Could not receive data"+                        close_action+                        return Nothing+                    ) :: IOProblem -> IO (Maybe B.ByteString) )+                ++            OnlyHeaders_H1PC headers leftovers -> do +                -- print "OnlyHeaders_H1PC"+                -- Ready for action...+                -- ATTENTION: Not use for pushed streams here....+                -- We must decide what to do if the user return those+                -- anyway.+                let +                    modified_headers = addExtraHeaders sessions_context headers+                (response_headers, _, data_and_conclusion) <- coherent_worker (+                        modified_headers, +                        Nothing+                    )+                (_, fragments) <- runConduit $ fuseBoth data_and_conclusion consume +                let +                    response_text =+                        serializeHTTPResponse response_headers fragments++                catch +                    (do+                        push_action response_text+                        return $ Just leftovers+                    )+                    ((\ _e -> do+                        debugM "Session.HTTP1" "Session abandoned"+                        close_action+                        return Nothing+                    ) :: IOProblem -> IO (Maybe B.ByteString) )++            HeadersAndBody_H1PC _headers _stopcondition _recv_leftovers -> do+                -- print "HeadersAndBody_H1PC"+                -- Let's see if I can go through the basic movements first, then through +                -- more complicated things.+                -- TODO: Implement posts and other requests with bodies....+                close_action+                error "NotImplemented requests with bodies"+++addExtraHeaders :: SessionsContext -> Headers -> Headers +addExtraHeaders sessions_context headers = +  let +    enriched_lens :: Lens' SessionsContext SessionsEnrichedHeaders+    enriched_lens = (sessionsConfig . sessionsEnrichedHeaders)+    -- Haskell laziness here!+    headers_editor = He.fromList headers +    -- TODO: Figure out which is the best way to put this contact in the +    --       source code+    protocol_lens = He.headerLens "second-transfer-eh--used-protocol"+    add_used_protocol = sessions_context ^. (enriched_lens . addUsedProtocol )+    he1 = if add_used_protocol +        then set protocol_lens (Just "HTTP/1.1") headers_editor+        else headers_editor+    result = He.toList he1 ++  in if add_used_protocol +        -- Nothing will be computed here if the headers are not modified.+        then result +        else headers
− hs-src/SecondTransfer/Http2/Framer.cpphs 
@@ -1,547 +0,0 @@--- The framer has two functions: to convert bytes to Frames and the other way around,--- and two keep track of flow-control quotas. -{-# LANGUAGE OverloadedStrings, StandaloneDeriving, FlexibleInstances, -             DeriveDataTypeable, TemplateHaskell #-}-{-# OPTIONS_HADDOCK hide #-}-module SecondTransfer.Http2.Framer (-    BadPrefaceException,--    wrapSession,-    http2FrameLength,--    -- Not needed anywhere, but supress the warning about unneeded symbol-    closeAction-    ) where ----import           Control.Concurrent-import           Control.Exception-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.Foldable                          (find)--import qualified Network.HTTP2                          as NH2--- Logging utilities-import           System.Log.Logger--import qualified Data.HashTable.IO                      as H--import           SecondTransfer.Sessions.Internal       (sessionExceptionHandler, nextSessionId, SessionsContext)-import           SecondTransfer.Http2.Session           -import           SecondTransfer.MainLoop.CoherentWorker (CoherentWorker)-import qualified SecondTransfer.MainLoop.Framer         as F-import           SecondTransfer.MainLoop.PushPullType   (Attendant, CloseAction,-                                                         PullAction, PushAction)-import           SecondTransfer.Utils                   (Word24, word24ToInt)-import           SecondTransfer.Exception---#include "Logging.cpphs"---http2PrefixLength :: Int-http2PrefixLength = B.length NH2.connectionPreface---- Let's do flow control here here .... --type HashTable k v = H.CuckooHashTable k v---type GlobalStreamId = Int---data FlowControlCommand = -     AddBytes_FCM Int ---- A hashtable from stream id to channel of availabiliy increases-type Stream2AvailSpace = HashTable GlobalStreamId (Chan FlowControlCommand)---data CanOutput = CanOutput---data NoHeadersInChannel = NoHeadersInChannel---data FramerSessionData = FramerSessionData {-      _stream2flow           :: Stream2AvailSpace-    , _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 -    }---L.makeLenses ''FramerSessionData---type FramerSession = ReaderT FramerSessionData IO---wrapSession :: CoherentWorker -> SessionsContext -> Attendant-wrapSession coherent_worker sessions_context push_action pull_action close_action = do--    let -        session_id_mvar = view nextSessionId sessions_context--    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-    last_stream_id            <- newMVar 0-    output_is_forbidden       <- newMVar False---    -- We need some shared state -    let framer_session_data = FramerSessionData {-        _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-        }---    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--        exc_handler :: Int -> SessionsContext -> FramerException -> IO ()-        exc_handler x y e = do-            modifyMVar_ output_is_forbidden (\ _ -> return True) -            sessionExceptionHandler Framer_HTTP2SessionComponent 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-    input_as_lbs = LB.fromStrict bs-  in -    Just $ (word24ToInt word24) + 9 -- Nine bytes that the frame header always uses-http2FrameLength _ = Nothing---addCapacity :: -        GlobalStreamId ->-        Int           -> -        FramerSession ()-addCapacity stream_id delta_cap = do --    if stream_id == 0 -      then -        -- TODO: Add session flow control-        return ()-      else do-        table <- view stream2flow  -        val <- liftIO $ H.lookup table stream_id-        case val of -            Nothing -> do-                -- ??-                liftIO $ putStrLn $ "Tried to update window of unexistent stream (creating): " ++ (show stream_id)-                (_, command_chan) <- startStreamOutputQueue stream_id -                -- And try again-                liftIO $ writeChan command_chan $ AddBytes_FCM delta_cap---            Just command_chan -> do-                liftIO $ writeChan command_chan $ AddBytes_FCM delta_cap-        ---- This works by pulling bytes from the input side of the pipeline and converting them to frames.--- The frames are then put in the SessionInput. In the other end of the SessionInput they can be --- interpreted according to their HTTP/2 meaning. --- --- This function also does part of the flow control: it registers WindowUpdate frames and triggers--- quota updates on the streams. -inputGatherer :: PullAction -> SessionInput -> FramerSession ()-inputGatherer pull_action session_input = do -    -- We can start by reading off the prefix....-    INSTRUMENTATION( liftIO $ debugM "HTTP2.Framer" "Entering InputGatherer" )-    (prefix, remaining) <- liftIO $ F.readLength http2PrefixLength pull_action--    if prefix /= NH2.connectionPreface -      then do -        sendGoAwayFrame NH2.ProtocolError-        liftIO $ do -            -- We just the the GoAway frame, although this is awfully early-            -- and probably wrong-            INSTRUMENTATION( errorM "HTTP2.Framer" "Invalid prologue")-            throwIO BadPrefaceException-      else -        INSTRUMENTATION( liftIO $ debugM "HTTP2.Framer" "Prologue validated" )-    let -        source::Source FramerSession B.ByteString-        source = transPipe liftIO $ F.readNextChunk http2FrameLength remaining pull_action-    ( source $$ consume)-  where --    consume :: Sink B.ByteString FramerSession ()-    consume = do -        maybe_bytes <- await -        -- Deserialize--        case maybe_bytes of --            Just bytes -> do-                let -                    error_or_frame = NH2.decodeFrame some_settings bytes-                    -- TODO: See how we can change these....-                    some_settings = NH2.defaultSettings--                case error_or_frame of --                    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 --                            (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 ()---                            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---                                        -- And set a new value -                                        liftIO $ putMVar old_default_stream_size_mvar new_default_stream_size---                                    Nothing -> -                                        -- This is a silenced internal error-                                        return ()--                                -- And send the frame down to the session, so that session specific settings-                                -- can be applied. -                                liftIO $ sendFrameToSession session_input frame---                            a_frame@(NH2.Frame (NH2.FrameHeader _ _ stream_id) _ )   -> do -                                -- Update the keep of last stream -                                lift $ updateLastStream $ NH2.fromStreamIdentifier stream_id--                                -- 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 --    -- We start by sending a settings frame -    pushFrame -        (NH2.EncodeInfo NH2.defaultFlags (NH2.toStreamIdentifier 0) Nothing)-        (NH2.SettingsFrame [])--    loopPart--  where ---    dataForFrame p1 p2 = -        LB.fromStrict $ NH2.encodeFrame p1 p2--    loopPart :: FramerSession ()-    loopPart = do --        command_or_frame  <- liftIO $ getFrameFromSession session_output--        case command_or_frame of --            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-                -- some circumstances... -                let stream_id = NH2.fromStreamIdentifier stream_idii-                s2o <- view stream2outputBytes-                lookup_result <- liftIO $ H.lookup s2o stream_id -                stream_bytes_chan <- case lookup_result of --                    Nothing ->  do -                        (bc, _) <- startStreamOutputQueue stream_id-                        return bc--                    Just bytes_chan -> return bytes_chan --                liftIO $ writeChan stream_bytes_chan $ dataForFrame p1 p2--                loopPart--            Right (p1, p2@(NH2.HeadersFrame _ _) ) -> do-                handleHeadersOfStream p1 p2-                -                loopPart--            Right (p1, p2@(NH2.ContinuationFrame _) ) -> do-                handleHeadersOfStream p1 p2--                loopPart--            Right (p1, p2) -> do -                -- Most other frames go right away... as long as no headers are in process...-                no_headers <- view noHeadersInChannel-                liftIO $ takeMVar no_headers-                pushFrame p1 p2 -                liftIO $ putMVar no_headers NoHeadersInChannel-                -                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--    --  -    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_HTTP2SessionComponent x y e---    read_state <- ask -    liftIO $ forkIO $ close_on_error session_id' sessions_context  $ runReaderT-        (flowControlOutput stream_id initial_cap "" command_chan bytes_chan)-        read_state--    return (bytes_chan , command_chan)----- 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-        -- Take it -        no_headers <- view noHeadersInChannel-        liftIO $ takeMVar no_headers-        pushFrame p1 frame_payload -        -- DONT PUT THE MvAR HERE --    | (frameIsHeaderOfStream frame_payload) && (frameEndsHeaders p1 frame_payload) = do-        no_headers <- view noHeadersInChannel-        liftIO $ takeMVar no_headers-        pushFrame p1 frame_payload -        -- Since we finish.... -        liftIO $ putMVar no_headers NoHeadersInChannel--    | 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-        liftIO $ putMVar no_headers NoHeadersInChannel-        return ()---frameIsHeaderOfStream :: NH2.FramePayload -> Bool-frameIsHeaderOfStream (NH2.HeadersFrame _  _ )-    = True-frameIsHeaderOfStream _                                       -    = False ---frameEndsHeaders  :: NH2.EncodeInfo -> NH2.FramePayload -> Bool -frameEndsHeaders (NH2.EncodeInfo flags _ _) (NH2.HeadersFrame _ _) = NH2.testEndHeader flags-frameEndsHeaders (NH2.EncodeInfo flags _ _) (NH2.ContinuationFrame _) = NH2.testEndHeader flags-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-    let bs = LB.fromStrict $ NH2.encodeFrame p1 p2  -    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` -            (C.bracket -                (takeMVar   can_output)-                (\ _ -> push_action bs)-                (\ c -> putMVar  can_output c)-            )----- 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 -    if leftovers == "" -      then do-        -- Get more data (possibly block waiting for it)-        bytes_to_send <- liftIO $ readChan bytes_chan-        flowControlOutput stream_id capacity  bytes_to_send commands_chan bytes_chan-      else do-        -- Length?-        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-            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)-            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 ()
+ hs-src/SecondTransfer/Http2/Framer.hs view
@@ -0,0 +1,562 @@+-- The framer has two functions: to convert bytes to Frames and the other way around,+-- and two keep track of flow-control quotas. +{-# LANGUAGE OverloadedStrings, StandaloneDeriving, FlexibleInstances, +             DeriveDataTypeable, TemplateHaskell #-}+{-# OPTIONS_HADDOCK hide #-}+module SecondTransfer.Http2.Framer (+    BadPrefaceException,++    wrapSession,+    http2FrameLength,++    -- Not needed anywhere, but supress the warning about unneeded symbol+    closeAction+    ) where ++++import           Control.Concurrent+import           Control.Exception+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.Foldable                          (find)++import qualified Network.HTTP2                          as NH2+-- Logging utilities+import           System.Log.Logger++import qualified Data.HashTable.IO                      as H++import           SecondTransfer.Sessions.Internal       (sessionExceptionHandler, nextSessionId, SessionsContext)+import           SecondTransfer.Http2.Session           +import           SecondTransfer.MainLoop.CoherentWorker (CoherentWorker)+import qualified SecondTransfer.MainLoop.Framer         as F+import           SecondTransfer.MainLoop.PushPullType   (Attendant, CloseAction,+                                                         PullAction, PushAction)+import           SecondTransfer.Utils                   (Word24, word24ToInt)+import           SecondTransfer.Exception+++#include "Logging.cpphs"+++http2PrefixLength :: Int+http2PrefixLength = B.length NH2.connectionPreface++-- Let's do flow control here here .... ++type HashTable k v = H.CuckooHashTable k v+++type GlobalStreamId = Int+++data FlowControlCommand = +     AddBytes_FCM Int ++-- A hashtable from stream id to channel of availabiliy increases+type Stream2AvailSpace = HashTable GlobalStreamId (Chan FlowControlCommand)+++data CanOutput = CanOutput+++data NoHeadersInChannel = NoHeadersInChannel+++data FramerSessionData = FramerSessionData {+      _stream2flow           :: Stream2AvailSpace+    , _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 +    }+++L.makeLenses ''FramerSessionData+++type FramerSession = ReaderT FramerSessionData IO+++wrapSession :: CoherentWorker -> SessionsContext -> Attendant+wrapSession coherent_worker sessions_context push_action pull_action close_action = do++    let +        session_id_mvar = view nextSessionId sessions_context++    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+    last_stream_id            <- newMVar 0+    output_is_forbidden       <- newMVar False+++    -- We need some shared state +    let framer_session_data = FramerSessionData {+        _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+        }+++    let +        -- TODO: Dodgy exception handling here...+        close_on_error session_id session_context comp = +            E.finally (+                E.catch(+                    (+                        E.catch+                            comp+                            (exc_handler session_id session_context)+                            )+                    )+                    (io_exc_handler session_id session_context)+                )+                close_action++        exc_handler :: Int -> SessionsContext -> FramerException -> IO ()+        exc_handler x y e = do+            modifyMVar_ output_is_forbidden (\ _ -> return True) +            errorM "HTTP2.Framer" "Exception went up"+            sessionExceptionHandler Framer_HTTP2SessionComponent x y e++        io_exc_handler :: Int -> SessionsContext -> IOProblem -> IO ()+        io_exc_handler x y e = do+            modifyMVar_ output_is_forbidden (\ _ -> return True) +            -- !!! These exceptions are way too common for we to care....+            -- errorM "HTTP2.Framer" "Exception went up"+            -- sessionExceptionHandler Framer_HTTP2SessionComponent 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+    input_as_lbs = LB.fromStrict bs+  in +    Just $ (word24ToInt word24) + 9 -- Nine bytes that the frame header always uses+http2FrameLength _ = Nothing+++addCapacity :: +        GlobalStreamId ->+        Int           -> +        FramerSession ()+addCapacity stream_id delta_cap = do ++    if stream_id == 0 +      then +        -- TODO: Add session flow control+        return ()+      else do+        table <- view stream2flow  +        val <- liftIO $ H.lookup table stream_id+        case val of +            Nothing -> do+                -- TODO: There is a very important compliance note here. Fix this, otherwise+                -- a peer can create a memory overflow by sending WindowUpdate on unexistent+                -- streams.+                liftIO $ putStrLn $ "Tried to update window of unexistent stream (creating): " ++ (show stream_id)+                (_, command_chan) <- startStreamOutputQueue stream_id +                -- And try again+                liftIO $ writeChan command_chan $ AddBytes_FCM delta_cap+++            Just command_chan -> do+                liftIO $ writeChan command_chan $ AddBytes_FCM delta_cap+        ++-- This works by pulling bytes from the input side of the pipeline and converting them to frames.+-- The frames are then put in the SessionInput. In the other end of the SessionInput they can be +-- interpreted according to their HTTP/2 meaning. +-- +-- This function also does part of the flow control: it registers WindowUpdate frames and triggers+-- quota updates on the streams. +inputGatherer :: PullAction -> SessionInput -> FramerSession ()+inputGatherer pull_action session_input = do +    -- We can start by reading off the prefix....+    (prefix, remaining) <- liftIO $ F.readLength http2PrefixLength pull_action++    if prefix /= NH2.connectionPreface +      then do +        sendGoAwayFrame NH2.ProtocolError+        liftIO $ do +            -- We just the the GoAway frame, although this is awfully early+            -- and probably wrong+            INSTRUMENTATION( errorM "HTTP2.Framer" "Invalid prologue")+            throwIO BadPrefaceException+      else +        INSTRUMENTATION( liftIO $ debugM "HTTP2.Framer" "Prologue validated" )+    let +        source::Source FramerSession B.ByteString+        source = transPipe liftIO $ F.readNextChunk http2FrameLength remaining pull_action+    ( source $$ consume)+  where ++    consume :: Sink B.ByteString FramerSession ()+    consume = do +        maybe_bytes <- await +        -- Deserialize++        case maybe_bytes of ++            Just bytes -> do+                let +                    error_or_frame = NH2.decodeFrame some_settings bytes+                    -- TODO: See how we can change these....+                    some_settings = NH2.defaultSettings++                case error_or_frame of ++                    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 ++                            (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 ()+++                            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+++                                        -- And set a new value +                                        liftIO $ putMVar old_default_stream_size_mvar new_default_stream_size+++                                    Nothing -> +                                        -- This is a silenced internal error+                                        return ()++                                -- And send the frame down to the session, so that session specific settings+                                -- can be applied. +                                liftIO $ sendFrameToSession session_input frame+++                            a_frame@(NH2.Frame (NH2.FrameHeader _ _ stream_id) _ )   -> do +                                -- Update the keep of last stream +                                lift $ updateLastStream $ NH2.fromStreamIdentifier stream_id++                                -- 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 ++    -- We start by sending a settings frame +    pushFrame +        (NH2.EncodeInfo NH2.defaultFlags (NH2.toStreamIdentifier 0) Nothing)+        (NH2.SettingsFrame [])++    loopPart++  where +++    dataForFrame p1 p2 = +        LB.fromStrict $ NH2.encodeFrame p1 p2++    loopPart :: FramerSession ()+    loopPart = do ++        command_or_frame  <- liftIO $ getFrameFromSession session_output++        case command_or_frame of ++            Left CancelSession_SOC -> do +                -- The session wants to cancel things+                INSTRUMENTATION( liftIO $ debugM "HTTP2.Framer" "CancelSession_SOC processed")+                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+                -- some circumstances... +                let stream_id = NH2.fromStreamIdentifier stream_idii+                s2o <- view stream2outputBytes+                lookup_result <- liftIO $ H.lookup s2o stream_id +                stream_bytes_chan <- case lookup_result of ++                    Nothing ->  do +                        (bc, _) <- startStreamOutputQueue stream_id+                        return bc++                    Just bytes_chan -> return bytes_chan ++                liftIO $ writeChan stream_bytes_chan $ dataForFrame p1 p2+                loopPart++            Right (p1, p2@(NH2.HeadersFrame _ _) ) -> do+                handleHeadersOfStream p1 p2+                loopPart++            Right (p1, p2@(NH2.ContinuationFrame _) ) -> do+                handleHeadersOfStream p1 p2+                loopPart++            Right (p1, p2) -> do +                -- Most other frames go right away... as long as no headers are in process...+                no_headers <- view noHeadersInChannel+                liftIO $ takeMVar no_headers+                pushFrame p1 p2 +                liftIO $ putMVar no_headers NoHeadersInChannel+                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++    --  +    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_HTTP2SessionComponent x y e+++    read_state <- ask +    liftIO $ forkIO $ close_on_error session_id' sessions_context  $ runReaderT+        (flowControlOutput stream_id initial_cap "" command_chan bytes_chan)+        read_state++    return (bytes_chan , command_chan)+++-- 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...+handleHeadersOfStream :: NH2.EncodeInfo -> NH2.FramePayload -> FramerSession ()+handleHeadersOfStream p1@(NH2.EncodeInfo _ _ _) frame_payload+    | (frameIsHeadersAndOpensStream frame_payload) && (not $ frameEndsHeaders p1 frame_payload) = do+        -- Take it +        no_headers <- view noHeadersInChannel+        liftIO $ takeMVar no_headers+        pushFrame p1 frame_payload +        -- DONT PUT THE MvAR HERE ++    | (frameIsHeadersAndOpensStream frame_payload) && (frameEndsHeaders p1 frame_payload) = do+        no_headers <- view noHeadersInChannel+        liftIO $ takeMVar no_headers+        pushFrame p1 frame_payload +        -- Since we finish.... +        liftIO $ putMVar no_headers NoHeadersInChannel++    | 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+        pushFrame p1 frame_payload+        liftIO $ putMVar no_headers NoHeadersInChannel+        return ()++    | otherwise = do  +        -- Nothing to do with the mvar, the no_headers should be empty+        pushFrame p1 frame_payload+++frameIsHeadersAndOpensStream :: NH2.FramePayload -> Bool+frameIsHeadersAndOpensStream (NH2.HeadersFrame _  _ )+    = True+frameIsHeadersAndOpensStream _                                       +    = False +++frameEndsHeaders  :: NH2.EncodeInfo -> NH2.FramePayload -> Bool +frameEndsHeaders (NH2.EncodeInfo flags _ _) (NH2.HeadersFrame _ _) = NH2.testEndHeader flags+frameEndsHeaders (NH2.EncodeInfo flags _ _) (NH2.ContinuationFrame _) = NH2.testEndHeader flags+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+    let bs = LB.fromStrict $ NH2.encodeFrame p1 p2  +    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` +            (C.bracket +                (takeMVar   can_output)+                (\ _ -> push_action bs)+                (\ c -> putMVar  can_output c)+            )+++-- 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 +    if leftovers == "" +      then do+        -- Get more data (possibly block waiting for it)+        bytes_to_send <- liftIO $ readChan bytes_chan+        flowControlOutput stream_id capacity  bytes_to_send commands_chan bytes_chan+      else do+        -- Length?+        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+            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)+            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 ()
− hs-src/SecondTransfer/Http2/Session.cpphs 
@@ -1,754 +0,0 @@--- Session: links frames to streams, and helps in ordering the header frames--- so that they don't get mixed with header frames from other streams when --- resources are being served concurrently.-{-# LANGUAGE FlexibleContexts, Rank2Types, TemplateHaskell, OverloadedStrings #-}-{-# OPTIONS_HADDOCK hide #-}-module SecondTransfer.Http2.Session(-    http2Session-    ,getFrameFromSession-    ,sendFrameToSession-    ,sendCommandToSession--    ,CoherentSession-    ,SessionInput(..)-    ,SessionInputCommand(..)-    ,SessionOutput(..)-    ,SessionOutputCommand(..)-    ,SessionCoordinates(..)-    ,SessionComponent(..)-    ,SessionsCallbacks(..)-    ,SessionsConfig(..)-    ,ErrorCallback--    -- Internal stuff-    ,OutputFrame-    ,InputFrame-    ) where--#include "Logging.cpphs"---- System grade utilities-import           Control.Concurrent                     (ThreadId, forkIO)-import           Control.Concurrent.Chan-import           Control.Exception                      (throwTo)-import qualified Control.Exception                      as E-import           Control.Monad                          (forever)-import           Control.Monad.IO.Class                 (liftIO)-import           Control.Monad.Trans.Reader--import           Control.Concurrent.MVar-import qualified Data.ByteString                        as B-import qualified Data.ByteString.Builder                as Bu-import qualified Data.ByteString.Lazy                   as Bl-import           Data.Conduit-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.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.Tokens-import           SecondTransfer.Sessions.Config-import           SecondTransfer.Sessions.Internal       (sessionExceptionHandler, SessionsContext)-import           SecondTransfer.Utils                   (unfoldChannelAndSource)-import           SecondTransfer.Exception---- Unfortunately the frame encoding API of Network.HTTP2 is a bit difficult to --- use :-( -type OutputFrame = (NH2.EncodeInfo, NH2.FramePayload)-type InputFrame  = NH2.Frame---useChunkLength :: Int -useChunkLength = 16384----- Singleton instance used for concurrency-data HeadersSent = HeadersSent ---- All streams put their data bits here. A "Nothing" value signals--- end of data. -type DataOutputToConveyor = (GlobalStreamId, Maybe B.ByteString)----- Whatever a worker thread is going to need comes here.... --- this is to make refactoring easier, but not strictly needed. -data WorkerThreadEnvironment = WorkerThreadEnvironment {-    -- What's the header stream id?-    _streamId :: GlobalStreamId--    -- A full block of headers can come here... the mvar in the middle should-    -- be populate to signal end of headers transmission. A thread will be suspended-    -- waiting for that-    , _headersOutput :: Chan (GlobalStreamId, MVar HeadersSent, Headers)--    -- And regular contents can come this way and thus be properly mixed-    -- with everything else.... for now... -    ,_dataOutput :: Chan DataOutputToConveyor--    ,_streamsCancelled_WTE :: MVar NS.IntSet--    }--makeLenses ''WorkerThreadEnvironment----- An HTTP/2 session. Basically a couple of channels ... -type Session = (SessionInput, SessionOutput)----- From outside, one can only write to this one ... the newtype is to enforce ---    this.-newtype SessionInput = SessionInput ( Chan (Either SessionInputCommand InputFrame) )-sendFrameToSession :: SessionInput  -> InputFrame -> IO ()-sendFrameToSession (SessionInput chan) frame = writeChan chan $ Right frame--sendCommandToSession :: SessionInput  -> SessionInputCommand -> IO ()-sendCommandToSession (SessionInput chan) command = writeChan chan $ Left command---- From outside, one can only read from this one -newtype SessionOutput = SessionOutput ( Chan (Either SessionOutputCommand OutputFrame) )-getFrameFromSession :: SessionOutput -> IO (Either SessionOutputCommand OutputFrame) -getFrameFromSession (SessionOutput chan) = readChan chan---type HashTable k v = H.CuckooHashTable k v----- Blaze builder could be more proper here... -type Stream2HeaderBlockFragment = HashTable GlobalStreamId Bu.Builder---type WorkerMonad = ReaderT WorkerThreadEnvironment IO ----- Have to figure out which are these...but I would expect to have things--- like unexpected aborts here in this type.-data SessionInputCommand = -    CancelSession_SIC-  deriving Show ----- temporary-data  SessionOutputCommand = -    CancelSession_SOC-  deriving Show----- 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 ---data PostInputMechanism = PostInputMechanism (Chan (Maybe B.ByteString), InputDataStream)----- NH2.Frame != Frame-data SessionData = SessionData {-    -- ATTENTION: Ignore the warning coming from here for now-    _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))--    -- Use to encode -    ,_toEncodeHeaders            :: MVar HP.DynamicTable-    -- And used to decode-    ,_toDecodeHeaders            :: MVar HP.DynamicTable--    -- Used for decoding the headers-    ,_stream2HeaderBlockFragment :: Stream2HeaderBlockFragment--    -- Used for worker threads... this is actually a pre-filled template-    -- I make copies of it in different contexts, and as needed. -    ,_forWorkerThread            :: WorkerThreadEnvironment--    ,_coherentWorker             :: CoherentWorker--    -- Some streams may be cancelled -    ,_streamsCancelled           :: MVar NS.IntSet--    -- Data input mechanism corresponding to some threads-    ,_stream2PostInputMechanism  :: HashTable Int PostInputMechanism --    -- 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. This way we get early finalization. -    ,_stream2WorkerThread        :: HashTable Int ThreadId--    -- Use to retrieve/set the session id-    ,_sessionIdAtSession         :: Int-    }---makeLenses ''SessionData-----                                v- {headers table size comes here!!}-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---    -- For incremental construction of headers...-    stream_request_headers    <- H.new :: IO Stream2HeaderBlockFragment--    -- Warning: we should find a way of coping with different table sizes.-    decode_headers_table      <- HP.newDynamicTableForDecoding 4096-    decode_headers_table_mvar <- newMVar decode_headers_table--    encode_headers_table      <- HP.newDynamicTableForEncoding 4096-    encode_headers_table_mvar <- newMVar encode_headers_table--    -- These ones need independent threads taking care of sending stuff-    -- their way... -    headers_output            <- newChan :: IO (Chan (GlobalStreamId, MVar HeadersSent, Headers))-    data_output               <- newChan :: IO (Chan DataOutputToConveyor)--    stream2postinputmechanism <- H.new -    stream2workerthread       <- H.new--    -- What about stream cancellation?-    cancelled_streams_mvar    <- newMVar $ NS.empty :: IO (MVar NS.IntSet)--    let for_worker_thread = WorkerThreadEnvironment {-        _streamId = error "NotInitialized"  -        ,_headersOutput = headers_output-        ,_dataOutput = data_output-        ,_streamsCancelled_WTE = cancelled_streams_mvar-        }--    let session_data  = SessionData {-        _sessionsContext             = sessions_context-        ,_sessionInput                = session_input -        ,_sessionOutput              = session_output_mvar-        ,_toDecodeHeaders            = decode_headers_table_mvar-        ,_toEncodeHeaders            = encode_headers_table_mvar-        ,_stream2HeaderBlockFragment = stream_request_headers-        ,_forWorkerThread            = for_worker_thread-        ,_coherentWorker             = coherent_worker-        ,_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 $ exc_guard SessionInputThread_HTTP2SessionComponent -           $ runReaderT sessionInputThread session_data- -    -- Create a thread that captures headers and sends them down the tube -    forkIO $ exc_guard SessionHeadersOutputThread_HTTP2SessionComponent -           $ runReaderT (headersOutputThread headers_output session_output_mvar) session_data--    -- Create a thread that captures data and sends it down the tube-    forkIO $ exc_guard SessionDataOutputThread_HTTP2SessionComponent -           $ dataOutputThread data_output session_output_mvar--    -- 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) )----- TODO: Some ill clients can break this thread with exceptions. Make these paths a bit---- more robust.-sessionInputThread :: ReaderT SessionData IO ()-sessionInputThread  = do -    INSTRUMENTATION( liftIO $ debugM "HTTP2.Session" "Entering sessionInputThread" )--    -- This is an introductory and declarative block... all of this is tail-executed-    -- every time that  a packet needs to be processed. It may be a good idea to abstract-    -- these values in a closure... -    session_input             <- view sessionInput -    -    decode_headers_table_mvar <- view toDecodeHeaders -    stream_request_headers    <- view stream2HeaderBlockFragment-    cancelled_streams_mvar    <- view streamsCancelled-    coherent_worker           <- view coherentWorker--    for_worker_thread_uns     <- view forWorkerThread-    stream2workerthread       <- view stream2WorkerThread--    input                     <- liftIO $ readChan session_input--    INSTRUMENTATION( liftIO $ debugM "HTTP2.Session" $ "Got a frame or a command: " ++ (show input) )--    case input of --        Left CancelSession_SIC -> do -            -- Good place to tear down worker threads... Let the rest of the finalization-            -- to the framer-            liftIO $ do -                H.mapM_-                    (\ (_, thread_id) -> do-                        throwTo thread_id StreamCancelledException-                        infoM "HTTP2.Session" $ "Stream successfully interrupted"-                    )-                    stream2workerthread--            -- We do not continue here, but instead let it finish-            return ()--        Right frame | Just (stream_id, bytes) <- frameIsHeaderOfStream frame -> do -            -- Just append the frames to streamRequestHeaders-            appendHeaderFragmentBlock stream_id bytes--            if frameEndsHeaders frame then -              do-                -- Let's decode the headers-                let for_worker_thread     = set streamId stream_id for_worker_thread_uns -                headers_bytes             <- getHeaderBytes stream_id-                dyn_table                 <- liftIO $ takeMVar decode_headers_table_mvar-                (new_table, header_list ) <- liftIO $ {-# SCC "decodeHeader" #-} HP.decodeHeader dyn_table headers_bytes-                -- Good moment to remove the headers from the table.... we don't want a space-                -- leak here -                liftIO $ do -                    H.delete stream_request_headers stream_id-                    putMVar decode_headers_table_mvar new_table--                -- TODO: Validate headers, abort session if the headers are invalid.-                -- Otherwise other invariants will break!!-                -- THIS IS PROBABLY THE BEST PLACE FOR DOING IT.--                -- If the headers end the request.... -                post_data_source <- if not (frameEndsStream frame)-                  then do -                    mechanism <- createMechanismForStream stream_id -                    let source = postDataSourceFromMechanism mechanism-                    return $ Just source-                  else do -                    return Nothing--                -- I'm clear to start the worker, in its own thread-                liftIO $ do -                    thread_id <- forkIO $ runReaderT -                        (workerThread (header_list, post_data_source) coherent_worker)-                        for_worker_thread -                    H.insert stream2workerthread stream_id thread_id--                return ()-            else -                -- Frame doesn't end the headers... it was added before... so-                -- probably do nothing -                return ()-                -            continue --        Right frame@(NH2.Frame _ (NH2.RSTStreamFrame _error_code_id)) -> do-            let stream_id = streamIdFromFrame frame-            liftIO $ do -                INSTRUMENTATION( infoM "HTTP2.Session" $ "Stream reset: " ++ (show _error_code_id) )-                cancelled_streams <- takeMVar cancelled_streams_mvar-                INSTRUMENTATION( infoM "HTTP2.Session" $ "Cancelled stream was: " ++ (show stream_id) )-                putMVar cancelled_streams_mvar $ NS.insert  stream_id cancelled_streams-                maybe_thread_id <- H.lookup stream2workerthread stream_id-                case maybe_thread_id  of -                    Nothing -> -                        -- This is actually more like an internal error-                        error "InterruptingUnexistentStream"--                    Just thread_id -> do-                        throwTo thread_id StreamCancelledException-                        INSTRUMENTATION( infoM "HTTP2.Session" $ "Stream successfully interrupted" )--            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-            let stream_id = NH2.fromStreamIdentifier nh2_stream_id-            -- TODO: Handle the cases where the stream_id doesn't match an already existent-            -- stream. In such cases it is justified to reset the connection with a  protocol_error.--            streamWorkerSendData stream_id somebytes-            -- After that data has been received and forwarded downstream, we can issue a windows update-            ---            -- TODO: We can use wider frames to avoid congestion...-            -- .... and we can also be more compositional with these short bursts of data....-            ---            -- TODO: Consider that the best place to output these frames can be somewhere else...-            ---            -- TODO: Use a special, with-quota queue here to do flow control. Don't send meaningless-            --       WindowUpdateFrame's-            sendOutFrame-                (NH2.EncodeInfo-                    NH2.defaultFlags-                    nh2_stream_id-                    Nothing-                )-                (NH2.WindowUpdateFrame-                    (fromIntegral (B.length somebytes))-                )-            sendOutFrame-                (NH2.EncodeInfo-                    NH2.defaultFlags-                    (NH2.toStreamIdentifier 0)-                    Nothing-                )-                (NH2.WindowUpdateFrame-                    (fromIntegral (B.length somebytes))-                )                --            if frameEndsStream frame  -              then do -                -- Good place to close the source ... -                closePostDataSource stream_id -              else -                return ()--            continue --        Right (NH2.Frame (NH2.FrameHeader _ flags _) (NH2.PingFrame _)) | NH2.testAck flags-> do -            -- Deal with pings: this is an Ack, so do nothing-            continue --        Right (NH2.Frame (NH2.FrameHeader _ _ _) (NH2.PingFrame somebytes))  -> do -            -- Deal with pings: NOT an Ack, so answer-            INSTRUMENTATION( liftIO $ debugM "HTTP2.Session" "Ping processed" )-            sendOutFrame-                (NH2.EncodeInfo-                    (NH2.setAck NH2.defaultFlags)-                    (NH2.toStreamIdentifier 0)-                    Nothing -                )-                (NH2.PingFrame somebytes)--            continue --        Right (NH2.Frame frame_header (NH2.SettingsFrame _)) | isSettingsAck frame_header -> do -            -- Frame was received by the peer, do nothing here...-            continue --        -- TODO: Do something with these settings!!-        Right (NH2.Frame _ (NH2.SettingsFrame settings_list))  -> do -            INSTRUMENTATION( liftIO $ debugM "HTTP2.Session" $ "Received settings: " ++ (show settings_list) )-            -- Just acknowledge the frame.... for now -            sendOutFrame -                (NH2.EncodeInfo-                    (NH2.setAck NH2.defaultFlags)-                    (NH2.toStreamIdentifier 0)-                    Nothing )-                (NH2.SettingsFrame [])--            continue ---        Right somethingelse -> do -            -- An undhandled case here....-            INSTRUMENTATION( liftIO $ errorM "HTTP2.Session" $  "Received problematic frame: " )-            liftIO $ errorM "HTTP2.Session" $  "..  " ++ (show somethingelse)--            continue --  where --    continue = sessionInputThread--    sendOutFrame :: NH2.EncodeInfo -> NH2.FramePayload -> ReaderT SessionData IO ()-    sendOutFrame encode_info payload = do -        session_output_mvar <- view sessionOutput --        session_output <- liftIO $ takeMVar session_output_mvar-        liftIO $ writeChan session_output $ Right (encode_info, payload)-        liftIO $ putMVar session_output_mvar session_output---frameEndsStream :: InputFrame -> Bool -frameEndsStream (NH2.Frame (NH2.FrameHeader _ flags _) _)  = NH2.testEndStream flags---createMechanismForStream :: GlobalStreamId -> ReaderT SessionData IO PostInputMechanism-createMechanismForStream stream_id = do -    (chan, source) <- liftIO $ unfoldChannelAndSource-    stream2postinputmechanism <- view stream2PostInputMechanism-    let pim = PostInputMechanism (chan, source)-    liftIO $ H.insert stream2postinputmechanism stream_id pim -    return pim----- TODO: Can be optimized by factoring out the mechanism lookup--- TODO IMPORTANT: This is a good place to drop the postinputmechanism--- for a stream, so that unprocessed data can be garbage-collected.-closePostDataSource :: GlobalStreamId -> ReaderT SessionData IO ()-closePostDataSource stream_id = do -    stream2postinputmechanism <- view stream2PostInputMechanism--    pim_maybe <- liftIO $ H.lookup stream2postinputmechanism stream_id --    case pim_maybe of --        Just (PostInputMechanism (chan, _))  -> -            liftIO $ writeChan chan Nothing--        Nothing -> -            -- TODO: This is a protocol error, handle it properly-            error "Internal error/closePostDataSource"---streamWorkerSendData :: Int -> B.ByteString -> ReaderT SessionData IO ()-streamWorkerSendData stream_id bytes = do -    s2pim <- view stream2PostInputMechanism-    pim_maybe <- liftIO $ H.lookup s2pim stream_id --    case pim_maybe of --        Just pim  -> -            sendBytesToPim pim bytes--        Nothing -> -            -- This is an internal error, the mechanism should be -            -- created when the headers end (and if the headers -            -- do not finish the stream)-            error "Internal error"---sendBytesToPim :: PostInputMechanism -> B.ByteString -> ReaderT SessionData IO ()-sendBytesToPim (PostInputMechanism (chan, _)) bytes = -    liftIO $ writeChan chan (Just bytes)---postDataSourceFromMechanism :: PostInputMechanism -> InputDataStream-postDataSourceFromMechanism (PostInputMechanism (_, source)) = source---isSettingsAck :: NH2.FrameHeader -> Bool -isSettingsAck (NH2.FrameHeader _ flags _) = -    NH2.testAck flags---isStreamCancelled :: GlobalStreamId  -> WorkerMonad Bool -isStreamCancelled stream_id = do -    cancelled_streams_mvar <- view streamsCancelled_WTE-    cancelled_streams <- liftIO $ readMVar cancelled_streams_mvar-    return $ NS.member stream_id cancelled_streams----workerThread :: Request -> CoherentWorker -> WorkerMonad ()-workerThread req coherent_worker =-  do-    headers_output <- view headersOutput-    stream_id      <- view streamId--    (headers, _, data_and_conclussion) <- liftIO $ coherent_worker req--    -- Now I send the headers, if that's possible at all-    headers_sent <- liftIO $ newEmptyMVar-    liftIO $ writeChan headers_output (stream_id, headers_sent, headers)--    -- At this moment I should ask if the stream hasn't been cancelled by the browser before-    -- commiting to the work of sending addtitional data-    is_stream_cancelled <- isStreamCancelled stream_id-    if not is_stream_cancelled--      then do-        -- I have a beautiful source that I can de-construct...-        -- TODO: Optionally pulling data out from a Conduit ....-        -- liftIO ( data_and_conclussion $$ (_sendDataOfStream stream_id) )-        -- -        -- This threadlet should block here waiting for the headers to finish going-        (maybe_footers, _) <- runConduit $-            (transPipe liftIO data_and_conclussion) -            `fuseBothMaybe` -            (sendDataOfStream stream_id headers_sent)-        -- BIG TODO: Send the footers ... likely stream conclusion semantics -        -- will need to be changed. -        return ()-      else --        return ()----                                                       v-- comp. monad.-sendDataOfStream :: GlobalStreamId -> MVar HeadersSent -> Sink B.ByteString (ReaderT WorkerThreadEnvironment IO) ()-sendDataOfStream stream_id headers_sent = do-    data_output <- view dataOutput-    -- Wait for all headers sent-    liftIO $ takeMVar headers_sent-    consumer data_output-  where -    consumer data_output = do -        maybe_bytes <- await -        case maybe_bytes of -            Nothing -> -                liftIO $ writeChan data_output (stream_id, Nothing)-            Just bytes -> do-                liftIO $ writeChan data_output (stream_id, Just bytes)-                consumer data_output---appendHeaderFragmentBlock :: GlobalStreamId -> B.ByteString -> ReaderT SessionData IO ()-appendHeaderFragmentBlock global_stream_id bytes = do -    ht <- view stream2HeaderBlockFragment -    maybe_old_block <- liftIO $ H.lookup ht global_stream_id-    new_block <- case maybe_old_block of --        Nothing -> do-            INSTRUMENTATION( liftIO $ infoM "HTTP2.Session" $ "Starting stream " ++ (show global_stream_id) )-            return $ Bu.byteString bytes--        Just something -> -            return $ something `mappend` (Bu.byteString bytes) --    liftIO $ H.insert ht global_stream_id new_block---getHeaderBytes :: GlobalStreamId -> ReaderT SessionData IO B.ByteString-getHeaderBytes global_stream_id = do -    ht <- view stream2HeaderBlockFragment -    Just bytes <- liftIO $ H.lookup ht global_stream_id-    return $ Bl.toStrict $ Bu.toLazyByteString bytes---frameIsHeaderOfStream :: InputFrame -> Maybe (GlobalStreamId, B.ByteString)-frameIsHeaderOfStream (NH2.Frame (NH2.FrameHeader _ _ stream_id) ( NH2.HeadersFrame _ block_fragment   ) )-    = Just (NH2.fromStreamIdentifier stream_id, block_fragment)-frameIsHeaderOfStream (NH2.Frame (NH2.FrameHeader _ _ stream_id) ( NH2.ContinuationFrame block_fragment) )-    = Just (NH2.fromStreamIdentifier stream_id, block_fragment)-frameIsHeaderOfStream _                                       -    = Nothing ---frameEndsHeaders  :: InputFrame -> Bool -frameEndsHeaders (NH2.Frame (NH2.FrameHeader _ flags _) _) = NH2.testEndHeader flags---streamIdFromFrame :: InputFrame -> GlobalStreamId-streamIdFromFrame (NH2.Frame (NH2.FrameHeader _ _ stream_id) _) = NH2.fromStreamIdentifier stream_id----- TODO: Have different size for the headers..... just now going with a default size of 16 k...--- TODO: Find a way to kill this thread....-headersOutputThread :: Chan (GlobalStreamId, MVar HeadersSent, Headers)-                       -> MVar (Chan (Either SessionOutputCommand OutputFrame)) -                       -> ReaderT SessionData IO ()-headersOutputThread input_chan session_output_mvar = forever $ do -    (stream_id, headers_ready_mvar, headers) <- liftIO $ readChan input_chan-    -- liftIO $ putStrLn $ "Output headers: " ++ (show headers)--    -- First encode the headers using the table-    encode_dyn_table_mvar <- view toEncodeHeaders--    encode_dyn_table <- liftIO $ takeMVar encode_dyn_table_mvar-    (new_dyn_table, data_to_send ) <- liftIO $ HP.encodeHeader HP.defaultEncodeStrategy encode_dyn_table headers-    liftIO $ putMVar encode_dyn_table_mvar new_dyn_table--    -- Now split the bytestring in chunks of the needed size.... -    bs_chunks <- return $! bytestringChunk useChunkLength data_to_send--    -- And send the chunks through while locking the output place....-    liftIO $ E.bracket-        (takeMVar session_output_mvar)-        (putMVar session_output_mvar )-        (\ session_output -> do-            writeIndividualHeaderFrames session_output stream_id bs_chunks True-            -- And say that the headers for this thread are out -            INSTRUMENTATION( infoM "HTTP2.Session" $ "Headers were output for stream " ++ (show stream_id) )-            putMVar headers_ready_mvar HeadersSent-            ) -  where -    writeIndividualHeaderFrames :: -        Chan (Either SessionOutputCommand OutputFrame)-        -> GlobalStreamId -        -> [B.ByteString] -        -> Bool -        -> IO ()-    writeIndividualHeaderFrames session_output stream_id (last_fragment:[]) is_first = -        writeChan session_output $ Right ( NH2.EncodeInfo {-            NH2.encodeFlags     = NH2.setEndHeader NH2.defaultFlags -            ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id -            ,NH2.encodePadding  = Nothing }, -            (if is_first then NH2.HeadersFrame Nothing last_fragment else  NH2.ContinuationFrame last_fragment)-            )-    writeIndividualHeaderFrames session_output stream_id  (fragment:xs) is_first = do -        writeChan session_output $ Right ( NH2.EncodeInfo {-            NH2.encodeFlags     = NH2.defaultFlags -            ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id -            ,NH2.encodePadding  = Nothing }, -            (if is_first then NH2.HeadersFrame Nothing fragment else  NH2.ContinuationFrame fragment)-            )-        writeIndividualHeaderFrames session_output stream_id xs False---bytestringChunk :: Int -> B.ByteString -> [B.ByteString]-bytestringChunk len s | (B.length s) < len = [ s ]-bytestringChunk len s = h:(bytestringChunk len xs)-  where -    (h, xs) = B.splitAt len s ----- TODO: find a clean way to finish this thread (maybe with negative stream ids?)--- TODO: This function does non-optimal chunking for the case where responses are---       actually streamed.... in those cases we need to keep state for frames in ---       some other format.... --- TODO: Right now, we are transmitting an empty last frame with the end-of-stream---       flag set. I'm afraid that the only---       way to avoid that is by holding a frame or by augmenting the end-user interface---       so that the user can signal which one is the last frame. The first approach---       restricts responsiviness, the second one clutters things.-dataOutputThread :: Chan DataOutputToConveyor-                    -> MVar (Chan (Either SessionOutputCommand OutputFrame)) -                    -> IO ()-dataOutputThread input_chan session_output_mvar = forever $ do -    (stream_id, maybe_contents) <- readChan input_chan-    case maybe_contents of -        Nothing -> do-            liftIO $ do-                INSTRUMENTATION( debugM "HTTP2.Session" "End-of-stream flag set " )-                withLockedSessionOutput-                    (\ session_output ->    writeChan session_output $ Right ( NH2.EncodeInfo {-                             NH2.encodeFlags     = NH2.setEndStream NH2.defaultFlags-                            ,NH2.encodeStreamId  = NH2.toStreamIdentifier stream_id -                            ,NH2.encodePadding   = Nothing }, -                            NH2.DataFrame ""-                            )-                        )--        Just contents -> do -            -- And now just simply output it...-            let bs_chunks = bytestringChunk useChunkLength $! contents-            -- And send the chunks through while locking the output place....-            writeContinuations bs_chunks stream_id-            -  where --    withLockedSessionOutput = E.bracket -        (takeMVar session_output_mvar) -        (putMVar session_output_mvar) -- <-- There is an implicit argument there!!--    writeContinuations :: [B.ByteString] -> GlobalStreamId  -> IO ()-    writeContinuations fragments stream_id  = mapM_ (\ fragment -> -        withLockedSessionOutput (\ session_output -> writeChan session_output $ Right ( NH2.EncodeInfo {-            NH2.encodeFlags     = NH2.defaultFlags -            ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id -            ,NH2.encodePadding  = Nothing }, -            NH2.DataFrame fragment ) )-        ) fragments--
+ hs-src/SecondTransfer/Http2/Session.hs view
@@ -0,0 +1,965 @@+-- Session: links frames to streams, and helps in ordering the header frames+-- so that they don't get mixed with header frames from other streams when +-- resources are being served concurrently.+{-# LANGUAGE FlexibleContexts, Rank2Types, TemplateHaskell, OverloadedStrings #-}+{-# OPTIONS_HADDOCK hide #-}+module SecondTransfer.Http2.Session(+    http2Session+    ,getFrameFromSession+    ,sendFrameToSession+    ,sendCommandToSession++    ,CoherentSession+    ,SessionInput(..)+    ,SessionInputCommand(..)+    ,SessionOutput(..)+    ,SessionOutputCommand(..)+    ,SessionCoordinates(..)+    ,SessionComponent(..)+    ,SessionsCallbacks(..)+    ,SessionsConfig(..)+    ,ErrorCallback++    -- Internal stuff+    ,OutputFrame+    ,InputFrame+    ) where++#include "Logging.cpphs"++-- System grade utilities+import           Control.Concurrent                     (ThreadId, forkIO)+import           Control.Concurrent.Chan+import           Control.Exception                      (throwTo)+import qualified Control.Exception                      as E+import           Control.Monad                          (forever)+import           Control.Monad.IO.Class                 (liftIO)+import           Control.Monad.Trans.Reader++import           Control.Concurrent.MVar+import qualified Data.ByteString                        as B+import qualified Data.ByteString.Builder                as Bu+import qualified Data.ByteString.Lazy                   as Bl+import           Data.Conduit+import qualified Data.HashTable.IO                      as H+import qualified Data.IntSet                            as NS+import           Data.Maybe                             (isJust)+#ifndef IMPLICIT_MONOID+import           Data.Monoid                            (mappend)+#endif++import           Control.Lens+ +-- No framing layer here... let's use Kazu's Yamamoto library+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.Tokens+import           SecondTransfer.Sessions.Config+import           SecondTransfer.Sessions.Internal       (sessionExceptionHandler, SessionsContext, sessionsConfig)+import           SecondTransfer.Utils                   (unfoldChannelAndSource)+import           SecondTransfer.Exception+import qualified SecondTransfer.Utils.HTTPHeaders       as He++-- Unfortunately the frame encoding API of Network.HTTP2 is a bit difficult to +-- use :-( +type OutputFrame = (NH2.EncodeInfo, NH2.FramePayload)+type InputFrame  = NH2.Frame+++useChunkLength :: Int +useChunkLength = 16384+++-- Singleton instance used for concurrency+data HeadersSent = HeadersSent ++-- All streams put their data bits here. A "Nothing" value signals+-- end of data. +type DataOutputToConveyor = (GlobalStreamId, Maybe B.ByteString)+++-- Whatever a worker thread is going to need comes here.... +-- this is to make refactoring easier, but not strictly needed. +data WorkerThreadEnvironment = WorkerThreadEnvironment {+    -- What's the header stream id?+    _streamId :: GlobalStreamId++    -- A full block of headers can come here... the mvar in the middle should+    -- be populate to signal end of headers transmission. A thread will be suspended+    -- waiting for that+    , _headersOutput :: Chan (GlobalStreamId, MVar HeadersSent, Headers)++    -- And regular contents can come this way and thus be properly mixed+    -- with everything else.... for now... +    ,_dataOutput :: Chan DataOutputToConveyor++    ,_streamsCancelled_WTE :: MVar NS.IntSet++    }++makeLenses ''WorkerThreadEnvironment+++-- An HTTP/2 session. Basically a couple of channels ... +type Session = (SessionInput, SessionOutput)+++-- From outside, one can only write to this one ... the newtype is to enforce +--    this.+newtype SessionInput = SessionInput ( Chan (Either SessionInputCommand InputFrame) )+sendFrameToSession :: SessionInput  -> InputFrame -> IO ()+sendFrameToSession (SessionInput chan) frame = writeChan chan $ Right frame++sendCommandToSession :: SessionInput  -> SessionInputCommand -> IO ()+sendCommandToSession (SessionInput chan) command = writeChan chan $ Left command++-- From outside, one can only read from this one +newtype SessionOutput = SessionOutput ( Chan (Either SessionOutputCommand OutputFrame) )+getFrameFromSession :: SessionOutput -> IO (Either SessionOutputCommand OutputFrame) +getFrameFromSession (SessionOutput chan) = readChan chan+++type HashTable k v = H.CuckooHashTable k v+++type Stream2HeaderBlockFragment = HashTable GlobalStreamId Bu.Builder+++type WorkerMonad = ReaderT WorkerThreadEnvironment IO +++-- Have to figure out which are these...but I would expect to have things+-- like unexpected aborts here in this type.+data SessionInputCommand = +    CancelSession_SIC+  deriving Show +++-- temporary+data  SessionOutputCommand = +    CancelSession_SOC+  deriving Show+++-- 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 +++data PostInputMechanism = PostInputMechanism (Chan (Maybe B.ByteString), InputDataStream)++-- Settings imposed by the peer+data SessionSettings = SessionSettings {+    _pushEnabled :: Bool +    }++makeLenses ''SessionSettings+++-- NH2.Frame != Frame+data SessionData = SessionData {+    -- ATTENTION: Ignore the warning coming from here for now+    _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))++    -- Use to encode +    ,_toEncodeHeaders            :: MVar HP.DynamicTable+    -- And used to decode+    ,_toDecodeHeaders            :: MVar HP.DynamicTable+    -- While I'm receiving headers, anything which +    -- is not a header should end in the connection being +    -- closed +    ,_receivingHeaders           :: MVar (Maybe Int)+    -- _lastGoodStream is used both to report the last good stream in the +    -- GoAwayFrame and to keep track of streams oppened by the client. In +    -- other words, it contains the stream_id of the last valid client +    -- stream and is updated as soon as the first frame of that stream is +    -- received.+    ,_lastGoodStream             :: MVar Int++    -- Used for decoding the headers+    ,_stream2HeaderBlockFragment :: Stream2HeaderBlockFragment++    -- Used for worker threads... this is actually a pre-filled template+    -- I make copies of it in different contexts, and as needed. +    ,_forWorkerThread            :: WorkerThreadEnvironment++    ,_coherentWorker             :: CoherentWorker++    -- Some streams may be cancelled +    ,_streamsCancelled           :: MVar NS.IntSet++    -- Data input mechanism corresponding to some threads+    ,_stream2PostInputMechanism  :: HashTable Int PostInputMechanism ++    -- 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. This way we get early finalization. +    ,_stream2WorkerThread        :: HashTable Int ThreadId++    -- Use to retrieve/set the session id+    ,_sessionIdAtSession         :: Int++    -- And used to keep peer session settings+    ,_sessionSettings            :: MVar SessionSettings+    }+++makeLenses ''SessionData+++--                                v- {headers table size comes here!!}+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+++    -- For incremental construction of headers...+    stream_request_headers    <- H.new :: IO Stream2HeaderBlockFragment++    -- Warning: we should find a way of coping with different table sizes.+    decode_headers_table      <- HP.newDynamicTableForDecoding 4096+    decode_headers_table_mvar <- newMVar decode_headers_table++    encode_headers_table      <- HP.newDynamicTableForEncoding 4096+    encode_headers_table_mvar <- newMVar encode_headers_table++    -- These ones need independent threads taking care of sending stuff+    -- their way... +    headers_output            <- newChan :: IO (Chan (GlobalStreamId, MVar HeadersSent, Headers))+    data_output               <- newChan :: IO (Chan DataOutputToConveyor)++    stream2postinputmechanism <- H.new +    stream2workerthread       <- H.new+    last_good_stream_mvar     <- newMVar (-1)++    receiving_headers         <- newMVar Nothing+    session_settings          <- newMVar $ SessionSettings { _pushEnabled = True }++    -- What about stream cancellation?+    cancelled_streams_mvar    <- newMVar $ NS.empty :: IO (MVar NS.IntSet)++    let for_worker_thread = WorkerThreadEnvironment {+        _streamId = error "NotInitialized"  +        ,_headersOutput = headers_output+        ,_dataOutput = data_output+        ,_streamsCancelled_WTE = cancelled_streams_mvar+        }++    let session_data  = SessionData {+        _sessionsContext             = sessions_context+        ,_sessionInput                = session_input +        ,_sessionOutput              = session_output_mvar+        ,_toDecodeHeaders            = decode_headers_table_mvar+        ,_toEncodeHeaders            = encode_headers_table_mvar+        ,_stream2HeaderBlockFragment = stream_request_headers+        ,_forWorkerThread            = for_worker_thread+        ,_coherentWorker             = coherent_worker+        ,_streamsCancelled           = cancelled_streams_mvar+        ,_stream2PostInputMechanism  = stream2postinputmechanism+        ,_stream2WorkerThread        = stream2workerthread+        ,_sessionIdAtSession         = session_id+        ,_receivingHeaders           = receiving_headers+        ,_sessionSettings            = session_settings+        ,_lastGoodStream             = last_good_stream_mvar+        }++    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 +            (\e -> do +                INSTRUMENTATION( errorM "HTTP2.Session" "Exception processed" )+                exc_handler component e +            )++    -- Create an input thread that decodes frames...+    forkIO $ exc_guard SessionInputThread_HTTP2SessionComponent +           $ runReaderT sessionInputThread session_data+ +    -- Create a thread that captures headers and sends them down the tube +    forkIO $ exc_guard SessionHeadersOutputThread_HTTP2SessionComponent +           $ runReaderT (headersOutputThread headers_output session_output_mvar) session_data++    -- Create a thread that captures data and sends it down the tube+    forkIO $ exc_guard SessionDataOutputThread_HTTP2SessionComponent +           $ dataOutputThread data_output session_output_mvar++    -- 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) )+++-- TODO: Some ill clients can break this thread with exceptions. Make these paths a bit+--- more robust.+sessionInputThread :: ReaderT SessionData IO ()+sessionInputThread  = do +    INSTRUMENTATION( liftIO $ debugM "HTTP2.Session" "Entering sessionInputThread" )++    -- This is an introductory and declarative block... all of this is tail-executed+    -- every time that  a packet needs to be processed. It may be a good idea to abstract+    -- these values in a closure... +    session_input             <- view sessionInput +    +    decode_headers_table_mvar <- view toDecodeHeaders +    stream_request_headers    <- view stream2HeaderBlockFragment+    cancelled_streams_mvar    <- view streamsCancelled+    coherent_worker           <- view coherentWorker++    for_worker_thread_uns     <- view forWorkerThread+    stream2workerthread       <- view stream2WorkerThread+    receiving_headers_mvar    <- view receivingHeaders+    last_good_stream_mvar     <- view lastGoodStream++    input                     <- liftIO $ readChan session_input++    -- INSTRUMENTATION( liftIO $ infoM "HTTP2.Session" $ "Got a frame or a command: " ++ (show input) )++    case input of ++        Left CancelSession_SIC -> do +            -- Good place to tear down worker threads... Let the rest of the finalization+            -- to the framer+            liftIO $ do +                H.mapM_+                    (\ (_, thread_id) -> do+                        throwTo thread_id StreamCancelledException+                        return ()+                    )+                    stream2workerthread++            -- We do not continue here, but instead let it finish+            return ()++        -- The block below will process both HEADERS and CONTINUATION frames. +        -- TODO: As it stands now, the server will happily start a new stream with +        -- a CONTINUATION frame instead of a HEADERS frame. That's against the +        -- protocol.+        Right frame | Just (stream_id, bytes) <- isAboutHeaders frame -> do +            -- Just append the frames to streamRequestHeaders+            opens_stream <- appendHeaderFragmentBlock stream_id bytes++            if opens_stream +              then do +                maybe_rcv_headers_of <- liftIO $ takeMVar receiving_headers_mvar+                case maybe_rcv_headers_of of +                  Just _ -> do +                    -- INSTRUMENTATION( liftIO $ errorM "HTTP2.Session" "headers being received")+                    -- Bad client, it is already sending headers +                    -- and trying to open another one+                    closeConnectionBecauseIsInvalid NH2.ProtocolError+                    -- An exception will be thrown above, so to not complicate+                    -- control flow here too much.+                  Nothing -> do +                    -- Signal that we are receiving headers now, for this stream+                    liftIO $ putMVar receiving_headers_mvar (Just stream_id)+                    -- And go to check if the stream id is valid+                    last_good_stream <- liftIO $ takeMVar last_good_stream_mvar+                    if (odd stream_id ) && (stream_id > last_good_stream) +                      then do +                        -- We are golden, set the new good stream+                        liftIO $ putMVar last_good_stream_mvar (stream_id)+                      else do +                        -- We are not golden+                        INSTRUMENTATION( liftIO $ errorM "HTTP2.Session" "Protocol error: bad stream id")+                        closeConnectionBecauseIsInvalid NH2.ProtocolError+              else do +                maybe_rcv_headers_of <- liftIO $ takeMVar receiving_headers_mvar+                case maybe_rcv_headers_of of +                    Just a_stream_id | a_stream_id == stream_id -> do +                        -- Nothing to complain about +                        liftIO $ putMVar receiving_headers_mvar maybe_rcv_headers_of++                    Nothing -> error "InternalError, this should be set"++            if frameEndsHeaders frame then +              do+                -- Ok, let it be known that we are not receiving more headers +                liftIO $ modifyMVar_ +                    receiving_headers_mvar+                    (\ _ -> return Nothing ) +                -- Let's decode the headers+                let for_worker_thread     = set streamId stream_id for_worker_thread_uns +                headers_bytes             <- getHeaderBytes stream_id+                dyn_table                 <- liftIO $ takeMVar decode_headers_table_mvar+                (new_table, header_list ) <- liftIO $ HP.decodeHeader dyn_table headers_bytes+                -- Good moment to remove the headers from the table.... we don't want a space+                -- leak here +                liftIO $ do +                    H.delete stream_request_headers stream_id+                    putMVar decode_headers_table_mvar new_table++                -- TODO: Validate headers, abort session if the headers are invalid.+                -- Otherwise other invariants will break!!+                -- THIS IS PROBABLY THE BEST PLACE FOR DOING IT.+                let +                    headers_editor = He.fromList header_list ++                maybe_good_headers_editor <- validateIncomingHeaders headers_editor++                good_headers <- case maybe_good_headers_editor of +                    Just yes_they_are_good -> return yes_they_are_good+                    Nothing -> closeConnectionBecauseIsInvalid NH2.ProtocolError++                -- Add any extra headers, on demand+                headers_extra_good      <- addExtraHeaders good_headers+                let +                    header_list_after = He.toList headers_extra_good+                -- liftIO $ putStrLn $ "header list after " ++ (show header_list_after)++                -- If the headers end the request.... +                post_data_source <- if not (frameEndsStream frame)+                  then do +                    mechanism <- createMechanismForStream stream_id +                    let source = postDataSourceFromMechanism mechanism+                    return $ Just source+                  else do +                    return Nothing++                -- TODO: Handle the cases where a request tries to send data +                -- even if the method doesn't allow for data.++                -- I'm clear to start the worker, in its own thread+                liftIO $ do +                    thread_id <- forkIO $ runReaderT +                        (workerThread (header_list_after, post_data_source) coherent_worker)+                        for_worker_thread +                    H.insert stream2workerthread stream_id thread_id++                return ()+            else +                -- Frame doesn't end the headers... it was added before... so+                -- probably do nothing +                return ()+                +            continue ++        Right frame@(NH2.Frame _ (NH2.RSTStreamFrame _error_code_id)) -> do+            let stream_id = streamIdFromFrame frame+            liftIO $ do +                INSTRUMENTATION( infoM "HTTP2.Session" $ "Stream reset: " ++ (show _error_code_id) )+                cancelled_streams <- takeMVar cancelled_streams_mvar+                INSTRUMENTATION( infoM "HTTP2.Session" $ "Cancelled stream was: " ++ (show stream_id) )+                putMVar cancelled_streams_mvar $ NS.insert  stream_id cancelled_streams+                maybe_thread_id <- H.lookup stream2workerthread stream_id+                case maybe_thread_id  of +                    Nothing -> +                        -- This is actually more like an internal error, when this +                        -- happend, cancell the session+                        error "InterruptingUnexistentStream"++                    Just thread_id -> do+                        INSTRUMENTATION( infoM "HTTP2.Session" $ "Stream successfully interrupted" )+                        throwTo thread_id StreamCancelledException++            continue ++        Right frame@(NH2.Frame (NH2.FrameHeader _ _ nh2_stream_id) (NH2.DataFrame somebytes)) +          -> unlessReceivingHeaders $ do +            -- So I got data to process+            -- TODO: Handle end of stream+            let stream_id = NH2.fromStreamIdentifier nh2_stream_id+            -- TODO: Handle the cases where the stream_id doesn't match an already existent+            -- stream. In such cases it is justified to reset the connection with a  protocol_error.++            streamWorkerSendData stream_id somebytes+            -- After that data has been received and forwarded downstream, we can issue a windows update+            --+            -- TODO: We can use wider frames to avoid congestion...+            -- .... and we can also be more compositional with these short bursts of data....+            --+            -- TODO: Consider that the best place to output these frames can be somewhere else...+            --+            -- TODO: Use a special, with-quota queue here to do flow control. Don't send meaningless+            --       WindowUpdateFrame's+            sendOutFrame+                (NH2.EncodeInfo+                    NH2.defaultFlags+                    nh2_stream_id+                    Nothing+                )+                (NH2.WindowUpdateFrame+                    (fromIntegral (B.length somebytes))+                )+            sendOutFrame+                (NH2.EncodeInfo+                    NH2.defaultFlags+                    (NH2.toStreamIdentifier 0)+                    Nothing+                )+                (NH2.WindowUpdateFrame+                    (fromIntegral (B.length somebytes))+                )                ++            if frameEndsStream frame  +              then do +                -- Good place to close the source ... +                closePostDataSource stream_id +              else +                return ()++            continue ++        Right (NH2.Frame (NH2.FrameHeader _ flags _) (NH2.PingFrame _)) | NH2.testAck flags-> do +            -- Deal with pings: this is an Ack, so do nothing+            continue ++        Right (NH2.Frame (NH2.FrameHeader _ _ _) (NH2.PingFrame somebytes))  -> do +            -- Deal with pings: NOT an Ack, so answer+            INSTRUMENTATION( liftIO $ debugM "HTTP2.Session" "Ping processed" )+            sendOutFrame+                (NH2.EncodeInfo+                    (NH2.setAck NH2.defaultFlags)+                    (NH2.toStreamIdentifier 0)+                    Nothing +                )+                (NH2.PingFrame somebytes)++            continue ++        Right (NH2.Frame frame_header (NH2.SettingsFrame _)) | isSettingsAck frame_header -> do +            -- Frame was received by the peer, do nothing here...+            continue ++        -- TODO: Do something with these settings!!+        Right (NH2.Frame _ (NH2.SettingsFrame settings_list))  -> do +            INSTRUMENTATION( liftIO $ debugM "HTTP2.Session" $ "Received settings: " ++ (show settings_list) )+            -- Just acknowledge the frame.... for now +            sendOutFrame +                (NH2.EncodeInfo+                    (NH2.setAck NH2.defaultFlags)+                    (NH2.toStreamIdentifier 0)+                    Nothing )+                (NH2.SettingsFrame [])++            continue +++        Right somethingelse ->  unlessReceivingHeaders $ do +            -- An undhandled case here....+            INSTRUMENTATION( liftIO $ errorM "HTTP2.Session" $  "Received problematic frame: " )+            liftIO $ errorM "HTTP2.Session" $  "..  " ++ (show somethingelse)++            continue ++  where +    continue = sessionInputThread+++sendOutFrame :: NH2.EncodeInfo -> NH2.FramePayload -> ReaderT SessionData IO ()+sendOutFrame encode_info payload = do +    session_output_mvar <- view sessionOutput ++    session_output <- liftIO $ takeMVar session_output_mvar+    liftIO $ writeChan session_output $ Right (encode_info, payload)+    liftIO $ putMVar session_output_mvar session_output+++-- TODO: This function, but using the headers editor, triggers +--       some renormalization of the header order. A good thing, if +--       I get that order well enough....+addExtraHeaders :: He.HeaderEditor -> ReaderT SessionData IO He.HeaderEditor+addExtraHeaders headers_editor = do+    let +        enriched_lens = (sessionsContext . sessionsConfig .sessionsEnrichedHeaders )+        -- TODO: Figure out which is the best way to put this contact in the +        --       source code+        protocol_lens = He.headerLens "second-transfer-eh--used-protocol"++    add_used_protocol <- view (enriched_lens . addUsedProtocol )++    -- liftIO $ putStrLn $ "AAA" ++ (show add_used_protocol)++    let +        he1 = if add_used_protocol +            then set protocol_lens (Just "HTTP/2") headers_editor+            else headers_editor++    if add_used_protocol +        -- Nothing will be computed here if the headers are not modified.+        then return he1+        else return headers_editor+++validateIncomingHeaders :: He.HeaderEditor -> ReaderT SessionData IO (Maybe He.HeaderEditor)+validateIncomingHeaders headers_editor = do +    -- Check that the headers block comes with all mandatory headers. +    -- Right now I'm not checking that they come in the mandatory order though...+    -- +    -- Notice that this function will transform a "host" header to an ":authority"+    -- one.+    let +        h1 = He.replaceHostByAuthority headers_editor+        -- Check that headers are lowercase+        headers_are_lowercase = He.headersAreLowercaseAtHeaderEditor headers_editor+        -- Check that we have mandatory headers +        maybe_authority = h1 ^. (He.headerLens ":authority")+        maybe_method    = h1 ^. (He.headerLens ":method")+        maybe_scheme    = h1 ^. (He.headerLens ":scheme")+        maybe_path      = h1 ^. (He.headerLens ":path")++    if +        (isJust maybe_authority) && +        (isJust maybe_method) && +        (isJust maybe_scheme) && +        (isJust maybe_path ) +        then +            return (Just h1)+        else +            return Nothing +++-- Sends a GO_AWAY frame and raises an exception, effectively terminating the input +-- thread of the session. +closeConnectionBecauseIsInvalid :: NH2.ErrorCodeId -> ReaderT SessionData IO a+closeConnectionBecauseIsInvalid error_code = do +    liftIO $ errorM "HTTP2.Session" "closeConnectionBecauseIsInvalid called!"+    last_good_stream_mvar <- view lastGoodStream+    last_good_stream <- liftIO $ takeMVar last_good_stream_mvar+    session_output_mvar <- view sessionOutput +    stream2workerthread <- view stream2WorkerThread+    sendOutFrame+        (NH2.EncodeInfo+            NH2.defaultFlags+            (NH2.toStreamIdentifier 0)+            Nothing+        )+        (NH2.GoAwayFrame+            (NH2.toStreamIdentifier last_good_stream)+            error_code+            ""+        ) +    +    liftIO $ do +        -- Close all active threads for this session+        H.mapM_+            ( \(_stream_id, thread_id) -> +                    throwTo thread_id StreamCancelledException+            )+            stream2workerthread++        -- Notify the framer that the session is closing, so +        -- that it stops accepting frames from connected sources +        -- (Streams?)+        session_output <- takeMVar session_output_mvar+        writeChan session_output $ Left CancelSession_SOC+        putMVar session_output_mvar session_output++        -- And unwind the input thread in the session, so that the +        -- exception handler runs.... +        E.throw HTTP2ProtocolException+++frameEndsStream :: InputFrame -> Bool +frameEndsStream (NH2.Frame (NH2.FrameHeader _ flags _) _)  = NH2.testEndStream flags+++-- Executes its argument, unless receiving +-- headers, in which case the connection is closed.+unlessReceivingHeaders :: ReaderT SessionData IO a -> ReaderT SessionData IO a+unlessReceivingHeaders comp = do +    receiving_headers_mvar    <- view receivingHeaders+    -- First check if we are receiving headers+    maybe_recv_headers <- liftIO $ readMVar receiving_headers_mvar+    if isJust maybe_recv_headers +      then +        -- So, this frame is highly illegal+        closeConnectionBecauseIsInvalid NH2.ProtocolError+      else +        comp+++createMechanismForStream :: GlobalStreamId -> ReaderT SessionData IO PostInputMechanism+createMechanismForStream stream_id = do +    (chan, source) <- liftIO $ unfoldChannelAndSource+    stream2postinputmechanism <- view stream2PostInputMechanism+    let pim = PostInputMechanism (chan, source)+    liftIO $ H.insert stream2postinputmechanism stream_id pim +    return pim+++-- TODO: Can be optimized by factoring out the mechanism lookup+-- TODO IMPORTANT: This is a good place to drop the postinputmechanism+-- for a stream, so that unprocessed data can be garbage-collected.+closePostDataSource :: GlobalStreamId -> ReaderT SessionData IO ()+closePostDataSource stream_id = do +    stream2postinputmechanism <- view stream2PostInputMechanism++    pim_maybe <- liftIO $ H.lookup stream2postinputmechanism stream_id ++    case pim_maybe of ++        Just (PostInputMechanism (chan, _))  -> +            liftIO $ writeChan chan Nothing++        Nothing -> +            -- TODO: This is a protocol error, handle it properly+            error "Internal error/closePostDataSource"+++streamWorkerSendData :: Int -> B.ByteString -> ReaderT SessionData IO ()+streamWorkerSendData stream_id bytes = do +    s2pim <- view stream2PostInputMechanism+    pim_maybe <- liftIO $ H.lookup s2pim stream_id ++    case pim_maybe of ++        Just pim  -> +            sendBytesToPim pim bytes++        Nothing -> +            -- This is an internal error, the mechanism should be +            -- created when the headers end (and if the headers +            -- do not finish the stream)+            error "Internal error"+++sendBytesToPim :: PostInputMechanism -> B.ByteString -> ReaderT SessionData IO ()+sendBytesToPim (PostInputMechanism (chan, _)) bytes = +    liftIO $ writeChan chan (Just bytes)+++postDataSourceFromMechanism :: PostInputMechanism -> InputDataStream+postDataSourceFromMechanism (PostInputMechanism (_, source)) = source+++isSettingsAck :: NH2.FrameHeader -> Bool +isSettingsAck (NH2.FrameHeader _ flags _) = +    NH2.testAck flags+++isStreamCancelled :: GlobalStreamId  -> WorkerMonad Bool +isStreamCancelled stream_id = do +    cancelled_streams_mvar <- view streamsCancelled_WTE+    cancelled_streams <- liftIO $ readMVar cancelled_streams_mvar+    return $ NS.member stream_id cancelled_streams++++workerThread :: Request -> CoherentWorker -> WorkerMonad ()+workerThread req coherent_worker =+  do+    headers_output <- view headersOutput+    stream_id      <- view streamId++    -- TODO: Handle exceptions here: what happens if the coherent worker+    --       throws an exception signaling that the request is ill-formed+    --       and should be dropped? That could happen in a couple of occassions,+    --       but really most cases should be handled here in this file...+    (headers, _, data_and_conclussion) <- liftIO $ coherent_worker req++    -- Now I send the headers, if that's possible at all+    headers_sent <- liftIO $ newEmptyMVar+    liftIO $ writeChan headers_output (stream_id, headers_sent, headers)++    -- At this moment I should ask if the stream hasn't been cancelled by the browser before+    -- commiting to the work of sending addtitional data+    is_stream_cancelled <- isStreamCancelled stream_id+    if not is_stream_cancelled++      then do+        -- I have a beautiful source that I can de-construct...+        -- TODO: Optionally pulling data out from a Conduit ....+        -- liftIO ( data_and_conclussion $$ (_sendDataOfStream stream_id) )+        -- +        -- This threadlet should block here waiting for the headers to finish going+        (maybe_footers, _) <- runConduit $+            (transPipe liftIO data_and_conclussion) +            `fuseBothMaybe` +            (sendDataOfStream stream_id headers_sent)+        -- BIG TODO: Send the footers ... likely stream conclusion semantics +        -- will need to be changed. +        return ()+      else ++        return ()++--                                                       v-- comp. monad.+sendDataOfStream :: GlobalStreamId -> MVar HeadersSent -> Sink B.ByteString (ReaderT WorkerThreadEnvironment IO) ()+sendDataOfStream stream_id headers_sent = do+    data_output <- view dataOutput+    -- Wait for all headers sent+    liftIO $ takeMVar headers_sent+    consumer data_output+  where +    consumer data_output = do +        maybe_bytes <- await +        case maybe_bytes of +            Nothing -> +                liftIO $ writeChan data_output (stream_id, Nothing)+            Just bytes -> do+                liftIO $ writeChan data_output (stream_id, Just bytes)+                consumer data_output+++-- Returns if the frame is the first in the stream+appendHeaderFragmentBlock :: GlobalStreamId -> B.ByteString -> ReaderT SessionData IO Bool+appendHeaderFragmentBlock global_stream_id bytes = do +    ht <- view stream2HeaderBlockFragment +    maybe_old_block <- liftIO $ H.lookup ht global_stream_id+    (new_block, new_stream) <- case maybe_old_block of ++        Nothing -> do+            -- TODO: Make the commented message below more informative+            -- INSTRUMENTATION( liftIO $ debugM "HTTP2.Session" $ "Starting stream " ++ (show global_stream_id) )+            return $ (Bu.byteString bytes, True)++        Just something -> +            return $ (something `mappend` (Bu.byteString bytes), False)++    liftIO $ H.insert ht global_stream_id new_block+    return new_stream++getHeaderBytes :: GlobalStreamId -> ReaderT SessionData IO B.ByteString+getHeaderBytes global_stream_id = do +    ht <- view stream2HeaderBlockFragment +    Just bytes <- liftIO $ H.lookup ht global_stream_id+    return $ Bl.toStrict $ Bu.toLazyByteString bytes+++isAboutHeaders :: InputFrame -> Maybe (GlobalStreamId, B.ByteString)+isAboutHeaders (NH2.Frame (NH2.FrameHeader _ _ stream_id) ( NH2.HeadersFrame _ block_fragment   ) )+    = Just (NH2.fromStreamIdentifier stream_id, block_fragment)+isAboutHeaders (NH2.Frame (NH2.FrameHeader _ _ stream_id) ( NH2.ContinuationFrame block_fragment) )+    = Just (NH2.fromStreamIdentifier stream_id, block_fragment)+isAboutHeaders _                                       +    = Nothing +++frameEndsHeaders  :: InputFrame -> Bool +frameEndsHeaders (NH2.Frame (NH2.FrameHeader _ flags _) _) = NH2.testEndHeader flags+++streamIdFromFrame :: InputFrame -> GlobalStreamId+streamIdFromFrame (NH2.Frame (NH2.FrameHeader _ _ stream_id) _) = NH2.fromStreamIdentifier stream_id+++-- TODO: Have different size for the headers..... just now going with a default size of 16 k...+-- TODO: Find a way to kill this thread....+headersOutputThread :: Chan (GlobalStreamId, MVar HeadersSent, Headers)+                       -> MVar (Chan (Either SessionOutputCommand OutputFrame)) +                       -> ReaderT SessionData IO ()+headersOutputThread input_chan session_output_mvar = forever $ do +    (stream_id, headers_ready_mvar, headers) <- liftIO $ readChan input_chan++    -- First encode the headers using the table+    encode_dyn_table_mvar <- view toEncodeHeaders++    encode_dyn_table <- liftIO $ takeMVar encode_dyn_table_mvar+    (new_dyn_table, data_to_send ) <- liftIO $ HP.encodeHeader HP.defaultEncodeStrategy encode_dyn_table headers+    liftIO $ putMVar encode_dyn_table_mvar new_dyn_table++    -- Now split the bytestring in chunks of the needed size.... +    bs_chunks <- return $! bytestringChunk useChunkLength data_to_send++    -- And send the chunks through while locking the output place....+    liftIO $ E.bracket+        (takeMVar session_output_mvar)+        (putMVar session_output_mvar )+        (\ session_output -> do+            writeIndividualHeaderFrames session_output stream_id bs_chunks True+            -- And say that the headers for this thread are out +            -- INSTRUMENTATION( debugM "HTTP2.Session" $ "Headers were output for stream " ++ (show stream_id) )+            putMVar headers_ready_mvar HeadersSent+            ) +  where +    writeIndividualHeaderFrames :: +        Chan (Either SessionOutputCommand OutputFrame)+        -> GlobalStreamId +        -> [B.ByteString] +        -> Bool +        -> IO ()+    writeIndividualHeaderFrames session_output stream_id (last_fragment:[]) is_first = +        writeChan session_output $ Right ( NH2.EncodeInfo {+            NH2.encodeFlags     = NH2.setEndHeader NH2.defaultFlags +            ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id +            ,NH2.encodePadding  = Nothing }, +            (if is_first then NH2.HeadersFrame Nothing last_fragment else  NH2.ContinuationFrame last_fragment)+            )+    writeIndividualHeaderFrames session_output stream_id  (fragment:xs) is_first = do +        writeChan session_output $ Right ( NH2.EncodeInfo {+            NH2.encodeFlags     = NH2.defaultFlags +            ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id +            ,NH2.encodePadding  = Nothing }, +            (if is_first then NH2.HeadersFrame Nothing fragment else  NH2.ContinuationFrame fragment)+            )+        writeIndividualHeaderFrames session_output stream_id xs False+++bytestringChunk :: Int -> B.ByteString -> [B.ByteString]+bytestringChunk len s | (B.length s) < len = [ s ]+bytestringChunk len s = h:(bytestringChunk len xs)+  where +    (h, xs) = B.splitAt len s +++-- TODO: find a clean way to finish this thread (maybe with negative stream ids?)+-- TODO: This function does non-optimal chunking for the case where responses are+--       actually streamed.... in those cases we need to keep state for frames in +--       some other format.... +-- TODO: Right now, we are transmitting an empty last frame with the end-of-stream+--       flag set. I'm afraid that the only+--       way to avoid that is by holding a frame or by augmenting the end-user interface+--       so that the user can signal which one is the last frame. The first approach+--       restricts responsiviness, the second one clutters things.+dataOutputThread :: Chan DataOutputToConveyor+                    -> MVar (Chan (Either SessionOutputCommand OutputFrame)) +                    -> IO ()+dataOutputThread input_chan session_output_mvar = forever $ do +    (stream_id, maybe_contents) <- readChan input_chan+    case maybe_contents of +        Nothing -> do+            liftIO $ do+                withLockedSessionOutput+                    (\ session_output ->    writeChan session_output $ Right ( NH2.EncodeInfo {+                             NH2.encodeFlags     = NH2.setEndStream NH2.defaultFlags+                            ,NH2.encodeStreamId  = NH2.toStreamIdentifier stream_id +                            ,NH2.encodePadding   = Nothing }, +                            NH2.DataFrame ""+                            )+                        )++        Just contents -> do +            -- And now just simply output it...+            let bs_chunks = bytestringChunk useChunkLength $! contents+            -- And send the chunks through while locking the output place....+            writeContinuations bs_chunks stream_id+            +  where ++    withLockedSessionOutput = E.bracket +        (takeMVar session_output_mvar) +        (putMVar session_output_mvar) -- <-- There is an implicit argument there!!++    writeContinuations :: [B.ByteString] -> GlobalStreamId  -> IO ()+    writeContinuations fragments stream_id  = mapM_ (\ fragment -> +        withLockedSessionOutput (\ session_output -> writeChan session_output $ Right ( NH2.EncodeInfo {+            NH2.encodeFlags     = NH2.defaultFlags +            ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id +            ,NH2.encodePadding  = Nothing }, +            NH2.DataFrame fragment ) )+        ) fragments++
hs-src/SecondTransfer/MainLoop/CoherentWorker.hs view
@@ -8,6 +8,7 @@  module SecondTransfer.MainLoop.CoherentWorker(     getHeaderFromFlatList+    , nullFooter      , HeaderName     , HeaderValue@@ -22,6 +23,7 @@     , PushedStream     , DataAndConclusion     , InputDataStream+     ) where   @@ -80,6 +82,14 @@ --   all the headers of a request. For GET requests that's the entire request --   basically, but for POST and PUT requests this is just before the data  --   starts arriving to the server. +--+--   It is important that you consume the data in the cases where there is an +--   input stream, otherwise the memory is lost for the duration of the request,+--   and a malicious client can use that.+--+--   Also, notice that when handling requests your worker can be interrupted with+--   an asynchronous exception of type 'StreamCancelledException', if the peer+--   cancels the stream type CoherentWorker = Request -> IO PrincipalStream  @@ -103,3 +113,18 @@          Nothing                -> Nothing   ++-- | If you want to skip the footers, i.e., they are empty, use this +--   function to convert an ordinary Source to a DataAndConclusion.+nullFooter :: Source IO B.ByteString -> DataAndConclusion+nullFooter s = s =$= go +  where +    go = do +        i <- await +        case i of +            Nothing -> +                return []++            Just ii -> do+                yield ii +                go
hs-src/SecondTransfer/MainLoop/Framer.hs view
@@ -18,7 +18,10 @@ import qualified Data.ByteString.Lazy      as LB import           Data.Conduit --- import           Data.Monoid               (mappend, mempty)+#ifndef IMPLICIT_MONOID+import           Data.Monoid               +#endif+   type Framer m =        LB.ByteString                        -- Input left overs
− hs-src/SecondTransfer/MainLoop/OpenSSL_TLS.cpphs 
@@ -1,365 +0,0 @@-{-# LANGUAGE ForeignFunctionInterface, OverloadedStrings,  DeriveDataTypeable #-}-{-# OPTIONS_HADDOCK hide #-}-module SecondTransfer.MainLoop.OpenSSL_TLS(-    tlsServeWithALPN-    ,tlsServeWithALPNAndFinishOnRequest-    -- ,tlsServeWithALPNOnce--    ,TLSLayerGenericProblem(..)-    ,FinishRequest(..)-    ) where ----import           Control.Monad-import           Control.Concurrent.MVar    -import           Control.Exception  -import qualified Control.Exception  as      E-import           Data.Foldable              (foldMap)-import           Data.Typeable              -import           Data.Monoid                ()-import           Foreign-import           Foreign.C--import qualified Data.ByteString            as B-import qualified Data.ByteString.Builder    as BB-import           Data.ByteString.Char8      (pack)-import qualified Data.ByteString.Lazy       as LB-import qualified Data.ByteString.Unsafe     as BU--import           System.Log.Logger--import           SecondTransfer.MainLoop.PushPullType-import           SecondTransfer.Exception--#include "Logging.cpphs"---- | Exceptions inheriting from `IOProblem`. This is thrown by the --- OpenSSL subsystem to signal that the connection was broken or that --- otherwise there was a problem at the SSL layer. -data TLSLayerGenericProblem = TLSLayerGenericProblem String-    deriving (Show, Typeable)---instance Exception TLSLayerGenericProblem where -    toException = toException . IOProblem -    fromException x = do -        IOProblem a <- fromException x -        cast a---data InterruptibleEither a b = -    Left_I a -    |Right_I b -    |Interrupted----- | Singleton type. Used in conjunction with an `MVar`. If the MVar is full, ---   the fuction `tlsServeWithALPNAndFinishOnRequest` knows that it should finish---   at its earliest convenience and call the `CloseAction` for any open sessions.-data FinishRequest = FinishRequest----- These names are absolutely improper....--- Session creator-data Connection_t  --- Session-data Wired_t--type Connection_Ptr = Ptr Connection_t -type Wired_Ptr = Ptr Wired_t----- Actually, this makes a listener for new connections--- connection_t* make_connection(char* certificate_filename, char* privkey_filename, char* hostname, int portno, ---     char* protocol_list, int protocol_list_len)-foreign import ccall "make_connection" makeConnection :: -    CString         -- cert filename-    -> CString      -- privkey_filename-    -> CString      -- hostname-    -> CInt         -- port-    -> Ptr CChar    -- protocol list-    -> CInt         -- protocol list length-    -> IO Connection_Ptr--allOk :: CInt -allOk = 0 --badHappened :: CInt -badHappened = 1 --timeoutReached :: CInt -timeoutReached = 3---- int wait_for_connection(connection_t* conn, wired_session_t** wired_session);-foreign import ccall "wait_for_connection" waitForConnection :: Connection_Ptr -> CInt -> Ptr Wired_Ptr -> IO CInt ---- int send_data(wired_session_t* ws, char* buffer, int buffer_size);-foreign import ccall "send_data" sendData :: Wired_Ptr -> Ptr CChar -> CInt -> IO CInt ---- int recv_data(wired_session_t* ws, char* inbuffer, int buffer_size, int* data_recvd);-foreign import ccall "recv_data" recvData :: Wired_Ptr -> Ptr CChar -> CInt -> Ptr CInt -> IO CInt---- int get_selected_protocol(wired_session_t* ws){ return ws->protocol_index; }-foreign import ccall "get_selected_protocol" getSelectedProtocol :: Wired_Ptr -> IO CInt---- void dispose_wired_session(wired_session_t* ws);-foreign import ccall "dispose_wired_session" disposeWiredSession :: Wired_Ptr -> IO ()--foreign import ccall "close_connection" closeConnection :: Connection_Ptr -> IO ()---useBufferSize :: Int-useBufferSize = 4096---type Protocols = [B.ByteString]---protocolsToWire :: Protocols -> B.ByteString-protocolsToWire protocols =  -    LB.toStrict . BB.toLazyByteString $ -        foldMap (\ protocol -                ->  (BB.lazyByteString . LB.fromChunks)-                    [ B.singleton $ fromIntegral $ B.length protocol,-                      protocol -                    ]-        ) protocols- ---- | Simple function to open -tlsServeWithALPN :: FilePath                -- ^ Path to a certificate the server is going to use to identify itself.-                                            --   Bear in mind that multiple domains can be served from the same HTTP/2 -                                            --   TLS socket, so please create the HTTP/2 certificate accordingly.-                                            --   Also, currently this function only accepts paths to certificates -                                            --   or certificate chains in .pem format. -                 -> FilePath                -- ^ Path to the key of your certificate. -                 -> String                  -- ^ Name of the network interface where you want to start your server-                 -> [(String, Attendant)]   -- ^ List of protocol names and the corresponding `Attendant` to use for -                                            --   each. This way you can serve both HTTP\/1.1 over TLS and HTTP\/2 in the-                                            --   same socket. When no ALPN negotiation is present during the negotiation, -                                            --   the first protocol in this list is used.-                 -> Int                     -- ^ Port to open to listen for connections. -                 -> IO ()-tlsServeWithALPN certificate_filename key_filename interface_name attendants interface_port = do --    let protocols_bs = protocolsToWire $ fmap (\ (s,_) -> pack s) attendants-    INSTRUMENTATION( infoM "OpenSSL" "Entering tlsServeWithALPN" )-    withCString certificate_filename $ \ c_certfn -> withCString key_filename $ \ c_keyfn -> withCString interface_name $ \ c_iname -> do --        connection_ptr <- BU.unsafeUseAsCStringLen protocols_bs $ \ (pchar, len) ->-            makeConnection -                c_certfn-                c_keyfn-                c_iname-                (fromIntegral interface_port)-                pchar -                (fromIntegral len)--        if connection_ptr == nullPtr -          then do-            INSTRUMENTATION( errorM "OpenSSL" "Could not create listening socket" )-            throwIO $ TLSLayerGenericProblem "Could not create listening end"-          else do-            INSTRUMENTATION( infoM "OpenSSL" "Listening socket created" )-            return ()--        forever $ do -            either_wired_ptr <- alloca $ \ wired_ptr_ptr -> -                let -                    tryOnce = do -                        result_code <- waitForConnection connection_ptr defaultWaitTime wired_ptr_ptr-                        let -                            r = case result_code of  -                                re  | re == allOk        -> do -                                        p <- peek wired_ptr_ptr-                                        INSTRUMENTATION( infoM "OpenSSL" "A connection was accepted" )-                                        return $ Right  p-                                    | re == timeoutReached -> tryOnce -                                    | re == badHappened  -> return $ Left ("A wait for connection failed" :: String)-                        r -                in tryOnce--            case either_wired_ptr of ---- Disable a warning-                Left _msg -> do -                    INSTRUMENTATION( errorM "OpenSSL" $ ".. wait for connection failed. " ++ _msg )-                    return ()--                Right wired_ptr -> do -                    already_closed_mvar <- newMVar False-                    let -                        pushAction datum = BU.unsafeUseAsCStringLen (LB.toStrict datum) $ \ (pchar, len) -> do -                            result <- sendData wired_ptr pchar (fromIntegral len)-                            case result of  -                                r | r == allOk           -> return ()-                                  | r == badHappened     -> throwIO $ TLSLayerGenericProblem "Could not send data"-                        pullAction = do -                            allocaBytes useBufferSize $ \ pcharbuffer -> -                                alloca $ \ data_recvd_ptr -> do -                                    result <- recvData wired_ptr pcharbuffer (fromIntegral useBufferSize) data_recvd_ptr-                                    INSTRUMENTATION( debugM "OpenSSL" "Received data" )-                                    recvd_bytes <- case result of -                                        r | r == allOk       -> peek data_recvd_ptr-                                          | r == badHappened -> throwIO $ TLSLayerGenericProblem "Could not receive data"--                                    B.packCStringLen (pcharbuffer, fromIntegral recvd_bytes)--                        -- Ensure that the socket and the struct are only closed once-                        closeAction = do-                            -- debugM "OpenSSL" "About to close connection"-                            b <- readMVar already_closed_mvar-                            if not b -                              then do-                                modifyMVar_ already_closed_mvar (\ _ -> return True)-                                disposeWiredSession wired_ptr-                                -- debugM "OpenSSL" "dispose clalled"-                              else -                                return ()--                    use_protocol <- getSelectedProtocol wired_ptr-                    INSTRUMENTATION( infoM "OpenSSL" $ "Selected protocol:" ++ (show use_protocol))--                    let -                        maybe_session_attendant = case fromIntegral use_protocol of -                            n | (use_protocol >= 0)  -> Just $ snd $ attendants !! n -                              -- Or just select the first one-                              | otherwise            -> Just . snd . head $ attendants--                    case maybe_session_attendant of --                        Just session_attendant -> -                            E.catch -                                (session_attendant pushAction pullAction closeAction)-                                ((\ e -> do -                                    INSTRUMENTATION( errorM "OpenSSL" " ** Session ended by TLSLayerGenericProblem (well handled)")-                                    throwIO e-                                )::TLSLayerGenericProblem -> IO () )---                        Nothing ->-                            return ()----- | Interruptible version of `tlsServeWithALPN`. Use the extra argument to ask ---   the server to finish: you pass an empty MVar and when you want to finish you ---   just populate it. -tlsServeWithALPNAndFinishOnRequest :: FilePath -                 -> FilePath              -- ^ Same as for `tlsServeWithALPN`             -                 -> String                -- ^ Same as for `tlsServeWithALPN`-                 -> [(String, Attendant)] -- ^ Same as for `tlsServeWithALPN`-                 -> Int                   -- ^ Same as for `tlsServeWithALPN`-                 -> MVar FinishRequest    -- ^ Finish request, write a value here to finish serving-                 -> IO ()-tlsServeWithALPNAndFinishOnRequest certificate_filename key_filename interface_name attendants interface_port finish_request = do --    let protocols_bs = protocolsToWire $ fmap (\ (s,_) -> pack s) attendants-    withCString certificate_filename $ \ c_certfn -> withCString key_filename $ \ c_keyfn -> withCString interface_name $ \ c_iname -> do --        -- Create an accepting endpoint-        connection_ptr <- BU.unsafeUseAsCStringLen protocols_bs $ \ (pchar, len) ->-            makeConnection -                c_certfn-                c_keyfn-                c_iname-                (fromIntegral interface_port)-                pchar -                (fromIntegral len)--        -- Create a computation that accepts a connection, runs a session on it and recurses-        let -            recursion = do -                -- Get a SSL session-                either_wired_ptr <- alloca $ \ wired_ptr_ptr -> -                    let -                        tryOnce = do -                            result_code <- waitForConnection connection_ptr smallWaitTime wired_ptr_ptr-                            let -                                r = case result_code of  -                                    re  | re == allOk        -> do -                                            p <- peek wired_ptr_ptr-                                            return $ Right_I  p-                                        | re == timeoutReached -> do -                                            got_finish_request <- tryTakeMVar finish_request-                                            case got_finish_request of -                                                Nothing ->-                                                    tryOnce-                                                Just _ ->-                                                    return Interrupted --                                        | re == badHappened  -> return $ Left_I "A wait for connection failed"-                            r -                    in tryOnce--                -- With the potentially obtained SSL session do...-                case either_wired_ptr of --                    Left_I msg -> do -                        errorM "OpenSSL" $ ".. wait for connection failed. " ++ msg--                        -- // .. //-                        recursion--                    Right_I wired_ptr -> do -                        already_closed_mvar <- newMVar False-                        let -                            pushAction datum = BU.unsafeUseAsCStringLen (LB.toStrict datum) $ \ (pchar, len) -> do -                                result <- sendData wired_ptr pchar (fromIntegral len)-                                case result of  -                                    r | r == allOk           -> return ()-                                      | r == badHappened     -> throwIO $ TLSLayerGenericProblem "Could not send data"-                            pullAction = do -                                allocaBytes useBufferSize $ \ pcharbuffer -> -                                    alloca $ \ data_recvd_ptr -> do -                                        result <- recvData wired_ptr pcharbuffer (fromIntegral useBufferSize) data_recvd_ptr-                                        recvd_bytes <- case result of -                                            r | r == allOk       -> peek data_recvd_ptr-                                              | r == badHappened -> throwIO $ TLSLayerGenericProblem "Could not receive data"--                                        B.packCStringLen (pcharbuffer, fromIntegral recvd_bytes)-                            closeAction = do-                                -- debugM "OpenSSL" "Close action about to be invoked"-                                b <- readMVar already_closed_mvar-                                if not b -                                  then do-                                    modifyMVar_ already_closed_mvar (\ _ -> return True)-                                    disposeWiredSession wired_ptr-                                    -- debugM "OpenSSL" "Close action invoked"-                                  else -                                    return ()--                        use_protocol <- getSelectedProtocol wired_ptr--                        infoM "OpenSSL" $ ".. Using protocol: " ++ (show use_protocol)--                        let -                            maybe_session_attendant = case fromIntegral use_protocol of -                                n | (use_protocol >= 0)  -> Just $ snd $ attendants !! n -                                  | otherwise            -> Just . snd . head $ attendants--                        case maybe_session_attendant of --                            Just session_attendant -> -                                session_attendant pushAction pullAction closeAction--                            Nothing ->-                                return ()--                        -- // .. //-                        recursion --                    Interrupted -> do-                        infoM "OpenSSL" "Connection closed"-                        closeConnection connection_ptr--        -- Start the loop defined above...-        recursion ---- When we are using the eternal version of this function, wake up --- each second .... -defaultWaitTime :: CInt-defaultWaitTime = 200000--- Okej, more responsiviness needed -smallWaitTime :: CInt -smallWaitTime = 50000
+ hs-src/SecondTransfer/MainLoop/OpenSSL_TLS.hs view
@@ -0,0 +1,350 @@+{-# LANGUAGE ForeignFunctionInterface, OverloadedStrings,  DeriveDataTypeable #-}+{-# OPTIONS_HADDOCK hide #-}+module SecondTransfer.MainLoop.OpenSSL_TLS(+    tlsServeWithALPN+    ,tlsServeWithALPNAndFinishOnRequest+    -- ,tlsServeWithALPNOnce++    ,TLSLayerGenericProblem(..)+    ,FinishRequest(..)+    ) where ++import           Control.Monad+import           Control.Concurrent.MVar    +import           Control.Exception  +import qualified Control.Exception  as      E+#ifndef IMPLICIT_APPLICATIVE_FOLDABLE+import           Data.Foldable              (foldMap)+#endif+import           Data.Typeable              +import           Data.Monoid                ()+import           Foreign+import           Foreign.C++import qualified Data.ByteString            as B+import qualified Data.ByteString.Builder    as BB+import           Data.ByteString.Char8      (pack)+import qualified Data.ByteString.Lazy       as LB+import qualified Data.ByteString.Unsafe     as BU++-- import           System.Log.Logger++import           SecondTransfer.MainLoop.PushPullType+import           SecondTransfer.Exception++#include "Logging.cpphs"++-- | Exceptions inheriting from `IOProblem`. This is thrown by the +-- OpenSSL subsystem to signal that the connection was broken or that +-- otherwise there was a problem at the SSL layer. +data TLSLayerGenericProblem = TLSLayerGenericProblem String+    deriving (Show, Typeable)+++instance Exception TLSLayerGenericProblem where +    toException = toException . IOProblem +    fromException x = do +        IOProblem a <- fromException x +        cast a+++data InterruptibleEither a b = +    Left_I a +    |Right_I b +    |Interrupted+++-- | Singleton type. Used in conjunction with an `MVar`. If the MVar is full, +--   the fuction `tlsServeWithALPNAndFinishOnRequest` knows that it should finish+--   at its earliest convenience and call the `CloseAction` for any open sessions.+data FinishRequest = FinishRequest+++-- These names are absolutely improper....+-- Session creator+data Connection_t  +-- Session+data Wired_t++type Connection_Ptr = Ptr Connection_t +type Wired_Ptr = Ptr Wired_t+++-- Actually, this makes a listener for new connections+-- connection_t* make_connection(char* certificate_filename, char* privkey_filename, char* hostname, int portno, +--     char* protocol_list, int protocol_list_len)+foreign import ccall "make_connection" makeConnection :: +    CString         -- cert filename+    -> CString      -- privkey_filename+    -> CString      -- hostname+    -> CInt         -- port+    -> Ptr CChar    -- protocol list+    -> CInt         -- protocol list length+    -> IO Connection_Ptr++allOk :: CInt +allOk = 0 ++badHappened :: CInt +badHappened = 1 ++timeoutReached :: CInt +timeoutReached = 3++-- int wait_for_connection(connection_t* conn, wired_session_t** wired_session);+foreign import ccall "wait_for_connection" waitForConnection :: Connection_Ptr -> CInt -> Ptr Wired_Ptr -> IO CInt ++-- int send_data(wired_session_t* ws, char* buffer, int buffer_size);+foreign import ccall "send_data" sendData :: Wired_Ptr -> Ptr CChar -> CInt -> IO CInt ++-- int recv_data(wired_session_t* ws, char* inbuffer, int buffer_size, int* data_recvd);+foreign import ccall "recv_data" recvData :: Wired_Ptr -> Ptr CChar -> CInt -> Ptr CInt -> IO CInt++-- int get_selected_protocol(wired_session_t* ws){ return ws->protocol_index; }+foreign import ccall "get_selected_protocol" getSelectedProtocol :: Wired_Ptr -> IO CInt++-- void dispose_wired_session(wired_session_t* ws);+foreign import ccall "dispose_wired_session" disposeWiredSession :: Wired_Ptr -> IO ()++foreign import ccall "close_connection" closeConnection :: Connection_Ptr -> IO ()+++useBufferSize :: Int+useBufferSize = 4096+++type Protocols = [B.ByteString]+++protocolsToWire :: Protocols -> B.ByteString+protocolsToWire protocols =  +    LB.toStrict . BB.toLazyByteString $ +        foldMap (\ protocol +                ->  (BB.lazyByteString . LB.fromChunks)+                    [ B.singleton $ fromIntegral $ B.length protocol,+                      protocol +                    ]+        ) protocols+ ++-- | Simple function to open +tlsServeWithALPN :: FilePath                -- ^ Path to a certificate the server is going to use to identify itself.+                                            --   Bear in mind that multiple domains can be served from the same HTTP/2 +                                            --   TLS socket, so please create the HTTP/2 certificate accordingly.+                                            --   Also, currently this function only accepts paths to certificates +                                            --   or certificate chains in .pem format. +                 -> FilePath                -- ^ Path to the key of your certificate. +                 -> String                  -- ^ Name of the network interface where you want to start your server+                 -> [(String, Attendant)]   -- ^ List of protocol names and the corresponding `Attendant` to use for +                                            --   each. This way you can serve both HTTP\/1.1 over TLS and HTTP\/2 in the+                                            --   same socket. When no ALPN negotiation is present during the negotiation, +                                            --   the first protocol in this list is used.+                 -> Int                     -- ^ Port to open to listen for connections. +                 -> IO ()+tlsServeWithALPN certificate_filename key_filename interface_name attendants interface_port = do ++    let protocols_bs = protocolsToWire $ fmap (\ (s,_) -> pack s) attendants+    withCString certificate_filename $ \ c_certfn -> withCString key_filename $ \ c_keyfn -> withCString interface_name $ \ c_iname -> do ++        connection_ptr <- BU.unsafeUseAsCStringLen protocols_bs $ \ (pchar, len) ->+            makeConnection +                c_certfn+                c_keyfn+                c_iname+                (fromIntegral interface_port)+                pchar +                (fromIntegral len)++        if connection_ptr == nullPtr +          then do+            throwIO $ TLSLayerGenericProblem "Could not create listening end"+          else do+            return ()++        forever $ do +            either_wired_ptr <- alloca $ \ wired_ptr_ptr -> +                let +                    tryOnce = do +                        result_code <- waitForConnection connection_ptr defaultWaitTime wired_ptr_ptr+                        let +                            r = case result_code of  +                                re  | re == allOk        -> do +                                        p <- peek wired_ptr_ptr+                                        return $ Right  p+                                    | re == timeoutReached -> tryOnce +                                    | re == badHappened  -> return $ Left ("A wait for connection failed" :: String)+                        r +                in tryOnce++            case either_wired_ptr of ++-- Disable a warning+                Left _msg -> do +                    return ()++                Right wired_ptr -> do ++                    (push_action, pull_action, close_action) <- provideActions wired_ptr ++                    use_protocol <- getSelectedProtocol wired_ptr++                    let +                        maybe_session_attendant = case fromIntegral use_protocol of +                            n | (use_protocol >= 0)  -> Just $ snd $ attendants !! n +                              -- Or just select the first one+                              | otherwise            -> Just . snd . head $ attendants++                    case maybe_session_attendant of ++                        Just session_attendant -> +                            E.catch +                                (session_attendant push_action pull_action close_action)+                                ((\ e -> do +                                    throwIO e+                                )::TLSLayerGenericProblem -> IO () )+++                        Nothing ->+                            return ()+++-- | Interruptible version of `tlsServeWithALPN`. Use the extra argument to ask +--   the server to finish: you pass an empty MVar and when you want to finish you +--   just populate it. +tlsServeWithALPNAndFinishOnRequest :: FilePath +                 -> FilePath              -- ^ Same as for `tlsServeWithALPN`             +                 -> String                -- ^ Same as for `tlsServeWithALPN`+                 -> [(String, Attendant)] -- ^ Same as for `tlsServeWithALPN`+                 -> Int                   -- ^ Same as for `tlsServeWithALPN`+                 -> MVar FinishRequest    -- ^ Finish request, write a value here to finish serving+                 -> IO ()+tlsServeWithALPNAndFinishOnRequest certificate_filename key_filename interface_name attendants interface_port finish_request = do ++    let protocols_bs = protocolsToWire $ fmap (\ (s,_) -> pack s) attendants+    withCString certificate_filename $ \ c_certfn -> withCString key_filename $ \ c_keyfn -> withCString interface_name $ \ c_iname -> do ++        -- Create an accepting endpoint+        connection_ptr <- BU.unsafeUseAsCStringLen protocols_bs $ \ (pchar, len) ->+            makeConnection +                c_certfn+                c_keyfn+                c_iname+                (fromIntegral interface_port)+                pchar +                (fromIntegral len)++        -- Create a computation that accepts a connection, runs a session on it and recurses+        let +            recursion = do +                -- Get a SSL session+                either_wired_ptr <- alloca $ \ wired_ptr_ptr -> +                    let +                        tryOnce = do +                            result_code <- waitForConnection connection_ptr smallWaitTime wired_ptr_ptr+                            let +                                r = case result_code of  +                                    re  | re == allOk        -> do +                                            p <- peek wired_ptr_ptr+                                            return $ Right_I  p+                                        | re == timeoutReached -> do +                                            got_finish_request <- tryTakeMVar finish_request+                                            case got_finish_request of +                                                Nothing ->+                                                    tryOnce+                                                Just _ ->+                                                    return Interrupted ++                                        | re == badHappened  -> return $ Left_I "A wait for connection failed"+                            r +                    in tryOnce++                -- With the potentially obtained SSL session do...+                case either_wired_ptr of ++                    Left_I _msg -> do +                        -- // .. //+                        recursion++                    Right_I wired_ptr -> do +                        (push_action, pull_action, close_action) <- provideActions wired_ptr ++                        use_protocol <- getSelectedProtocol wired_ptr++                        let +                            maybe_session_attendant = case fromIntegral use_protocol of +                                n | (use_protocol >= 0)  -> Just $ snd $ attendants !! n +                                  | otherwise            -> Just . snd . head $ attendants++                        case maybe_session_attendant of ++                            Just session_attendant -> +                                session_attendant push_action pull_action close_action++                            Nothing ->+                                return ()++                        -- // .. //+                        recursion ++                    Interrupted -> do+                        closeConnection connection_ptr++        -- Start the loop defined above...+        recursion ++-- When we are using the eternal version of this function, wake up +-- each second .... +defaultWaitTime :: CInt+defaultWaitTime = 200000+-- Okej, more responsiviness needed +smallWaitTime :: CInt +smallWaitTime = 50000++provideActions :: Wired_Ptr -> IO (LB.ByteString -> IO (), IO B.ByteString, IO ())+provideActions wired_ptr = do+    already_closed_mvar <- newMVar False+    let+        pushAction :: LB.ByteString -> IO ()+        pushAction datum = do +            already_closed <- readMVar already_closed_mvar+            if already_closed +              then do+                throwIO $ TLSLayerGenericProblem "Tried to send data on closed handle"+              else do+                BU.unsafeUseAsCStringLen (LB.toStrict datum) $ \ (pchar, len) -> do +                    result <- sendData wired_ptr pchar (fromIntegral len)+                    case result of  +                        r | r == allOk           -> do +                                return ()+                          | r == badHappened     -> do +                                throwIO $ TLSLayerGenericProblem "Could not send data"++        pullAction :: IO B.ByteString+        pullAction =  do+            already_closed <- readMVar already_closed_mvar+            if already_closed +              then do +                throwIO $ TLSLayerGenericProblem "Tried to receive on closed handle"+              else +                allocaBytes useBufferSize $ \ pcharbuffer -> +                    alloca $ \ data_recvd_ptr -> do +                        result <- recvData wired_ptr pcharbuffer (fromIntegral useBufferSize) data_recvd_ptr+                        recvd_bytes <- case result of +                            r | r == allOk       -> peek data_recvd_ptr+                              | r == badHappened -> do +                                    throwIO $ TLSLayerGenericProblem "Could not receive data"++                        B.packCStringLen (pcharbuffer, fromIntegral recvd_bytes)++        closeAction :: IO ()+        -- Ensure that the socket and the struct are only closed once+        closeAction = do+            b <- readMVar already_closed_mvar+            if not b +              then do+                modifyMVar_ already_closed_mvar (\ _ -> return True)+                disposeWiredSession wired_ptr+                -- debugM "OpenSSL" "dispose clalled"+              else +                return ()+    return (pushAction, pullAction, closeAction)
hs-src/SecondTransfer/Sessions/Config.hs view
@@ -1,14 +1,24 @@ {-# LANGUAGE FlexibleContexts, Rank2Types, TemplateHaskell, OverloadedStrings #-}+{- | Configuration and settings for the server. All constructor names are +     exported, but notice that they start with an underscore. +     They also have an equivalent lens without the +     underscore. Please prefer to use the lens interface.+-} module SecondTransfer.Sessions.Config(     sessionId     ,defaultSessionsConfig+    ,defaultSessionsEnrichedHeaders     ,sessionsCallbacks+    ,sessionsEnrichedHeaders     ,reportErrorCallback+    ,addUsedProtocol       ,SessionComponent(..)     ,SessionCoordinates(..)     ,SessionsCallbacks(..)+    ,SessionsEnrichedHeaders(..)+    -- ,UsedProtocol(..)     ,SessionsConfig(..)     ,ErrorCallback     ) where @@ -16,7 +26,7 @@  -- import           Control.Concurrent.MVar (MVar) import           Control.Exception       (SomeException)-import           Control.Lens            (Lens', makeLenses)+import           Control.Lens            (makeLenses)   -- | Information used to identify a particular session. @@ -38,7 +48,6 @@     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@@ -51,6 +60,10 @@     deriving Show  +-- Which protocol a session is using... no need for this right now+-- data UsedProtocol = +--      HTTP11_UsP+--     |HTTP2_UsP  -- | Used by this session engine to report an error at some component, in a particular --   session. @@ -66,20 +79,40 @@ makeLenses ''SessionsCallbacks  +-- | This is a temporal interface, but an useful one nonetheless. +--   By setting some values here to True, second-transfer will add+--   some headers to inbound requests, and some headers to outbound +--   requests. +data SessionsEnrichedHeaders = SessionsEnrichedHeaders {+    -- | Adds a second-transfer-eh--used-protocol header+    --   to inbound requests. Default: False+    _addUsedProtocol :: Bool+    }++makeLenses ''SessionsEnrichedHeaders++-- | Don't insert any extra-headers by default. +defaultSessionsEnrichedHeaders :: SessionsEnrichedHeaders+defaultSessionsEnrichedHeaders = SessionsEnrichedHeaders {+    _addUsedProtocol = False+    }++ -- | Configuration information you can provide to the session maker. data SessionsConfig = SessionsConfig {     -- | Session callbacks     _sessionsCallbacks :: SessionsCallbacks+    ,_sessionsEnrichedHeaders :: SessionsEnrichedHeaders } --- makeLenses ''SessionsConfig+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)+-- -- | Lens to access sessionsCallbacks in the `SessionsConfig` object.+-- sessionsCallbacks :: Lens' SessionsConfig SessionsCallbacks+-- sessionsCallbacks  f (+--     SessionsConfig {+--         _sessionsCallbacks= s +--     }) = fmap (\ s' -> SessionsConfig {_sessionsCallbacks = s'}) (f s)   -- | Creates a default sessions context. Modify as needed using @@ -88,6 +121,7 @@ defaultSessionsConfig = SessionsConfig {     _sessionsCallbacks = SessionsCallbacks {             _reportErrorCallback = Nothing-        }+        },+    _sessionsEnrichedHeaders = defaultSessionsEnrichedHeaders     }
+ hs-src/SecondTransfer/Utils/DevNull.hs view
@@ -0,0 +1,24 @@++module SecondTransfer.Utils.DevNull(+    dropIncomingData+    ) where +++import Control.Concurrent (forkIO)+import Data.Conduit ++import SecondTransfer.MainLoop.CoherentWorker ++-- TODO: Handling unnecessary data should be done in some other, less +-- harmfull way... need to think about that.++-- | If you are not processing the potential POST input in a request,+-- use this consumer to drop the data to oblivion. Otherwise it will +-- remain in an internal queue until the client closes the +-- stream, and if the client doesn't want to do so....+dropIncomingData :: Maybe InputDataStream -> IO ()+dropIncomingData Nothing = return ()+dropIncomingData (Just data_source) = do+    forkIO $ +        data_source $$ (awaitForever (\ _ -> return () ) )+    return ()
hs-src/SecondTransfer/Utils/HTTPHeaders.hs view
@@ -10,7 +10,8 @@     -- | These transformations are simple enough that don't require     --   going away from the list representation (see type `Headers`)      lowercaseHeaders-    ,headersAreValidHTTP2+    ,headersAreLowercase+    ,headersAreLowercaseAtHeaderEditor     ,fetchHeader     -- * Transformations based on maps     --@@ -44,9 +45,11 @@ import           Data.Word                              (Word8)  import           Data.Time.Format                       (formatTime, defaultTimeLocale)-import           Data.Time.Clock                        (UTCTime,getCurrentTime)+import           Data.Time.Clock                        (getCurrentTime) +#ifndef IMPLICIT_MONOID import           Control.Applicative                    ((<$>))+#endif  import           SecondTransfer.MainLoop.CoherentWorker (Headers) @@ -63,15 +66,22 @@   -- | Checks that headers are lowercase-headersAreValidHTTP2 :: Headers -> Bool -headersAreValidHTTP2 headers = -  let -    isOk a_header = not . T.any isUpper . decodeUtf8 . fst $ a_header-  in +headersAreLowercase :: Headers -> Bool +headersAreLowercase headers =      foldl-        (\ prev e -> (flip (&&)) (isOk  e) $! prev)+        (\ prev (hn, _) -> (flip (&&)) (aTitleIsLowercase  hn) $! prev)         True         headers++headersAreLowercaseAtHeaderEditor :: HeaderEditor -> Bool +headersAreLowercaseAtHeaderEditor header_editor = +    Ms.foldlWithKey'+        (\ prev hn _ -> (flip (&&)) (aTitleIsLowercase . toFlatBs $ hn) $! prev)+        True+        (innerMap header_editor)++aTitleIsLowercase :: B.ByteString -> Bool +aTitleIsLowercase a_title = not . T.any isUpper . decodeUtf8 $ a_title   -- | Looks for a given header
second-transfer.cabal view
@@ -7,7 +7,7 @@ -- PVP       summary:      +-+------- breaking API changes --                         | | +----- non-breaking API additions --                         | | | +--- code changes with no API change-version     :              0.5.2.2+version     :              0.5.3.1  synopsis    :              Second Transfer HTTP/2 web server @@ -53,7 +53,7 @@ source-repository this   type:     git   location: git@github.com:alcidesv/second-transfer.git-  tag:      0.5.2.2+  tag:      0.5.3.1  library @@ -64,6 +64,7 @@                   , SecondTransfer.Exception                   , SecondTransfer.Types                   , SecondTransfer.Utils.HTTPHeaders+                  , SecondTransfer.Utils.DevNull                   -- These are really internal modules, but  are exposed                    -- here for the sake of the test suite. They are hidden                    -- from the documentation.@@ -91,6 +92,8 @@    build-tools: cpphs +  default-extensions: CPP+   if flag(debug)     CPP-Options: -DENABLE_DEBUG     if !os(windows)@@ -98,6 +101,9 @@     else       CC-Options: "-DNDEBUG" +  if impl(ghc >= 7.10)+    CPP-Options: -DIMPLICIT_MONOID -DIMPLICIT_APPLICATIVE_FOLDABLE+   -- LANGUAGE extensions used by modules in this package.   -- other-extensions:       @@ -137,10 +143,12 @@   -- cc-options: -fPIC -pthread -g -O0   if flag(debug)       cc-options: -O0 -g3+      ld-options: -g3    -- cc-options: -g3 -O0 -  -- ghc-options: -O2+  -- ghc-options: -O2 -cpp  -pgmPcpphs  -optP--cpp+  ghc-options: -pgmPcpphs  -optP--cpp     extra-libraries: ssl crypto
tests/tests-hs-src/compiling_ok.hs view
@@ -6,6 +6,7 @@     , tlsServeWithALPNAndFinishOnRequest     , http2Attendant     , http11Attendant+    , dropIncomingData     , FinishRequest(..)     ) import SecondTransfer.Sessions(@@ -25,13 +26,15 @@   helloWorldWorker :: CoherentWorker-helloWorldWorker request = return (-    [-        (":status", "200")-    ],-    [], -- No pushed streams-    saysHello-    )+helloWorldWorker (_request_headers, _maybe_post_data) = do +    dropIncomingData _maybe_post_data+    return (+        [+            (":status", "200")+        ],+        [], -- No pushed streams+        saysHello+        )   -- For this program to work, it should be run from the top of