second-transfer 0.6.1.0 → 0.7.1.0

raw patch · 22 files changed

+1990/−393 lines, 22 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• README.md +5/−19
• cbits/tlsinc.c +45/−6
• changelog.md +5/−0
• hs-src/SecondTransfer.hs +7/−6
• hs-src/SecondTransfer/Exception.hs +21/−0
• hs-src/SecondTransfer/Http1/Session.hs +4/−4
• hs-src/SecondTransfer/Http2/Framer.hs +164/−59
• hs-src/SecondTransfer/Http2/MakeAttendant.hs +6/−10
• hs-src/SecondTransfer/Http2/Session.hs +708/−179
• hs-src/SecondTransfer/Http2/SimpleClient.hs +35/−0
• hs-src/SecondTransfer/MainLoop/ClientPetitioner.hs +15/−0
• hs-src/SecondTransfer/MainLoop/CoherentWorker.hs +21/−4
• hs-src/SecondTransfer/MainLoop/OpenSSL_TLS.hs +6/−6
• hs-src/SecondTransfer/MainLoop/PushPullType.hs +13/−12
• hs-src/SecondTransfer/Sessions/Config.hs +1/−0
• hs-src/SecondTransfer/Sessions/Internal.hs +4/−3
• second-transfer.cabal +18/−10
• tests/tests-hs-src/SecondTransfer/Test/DecoySession.hs +198/−60
• tests/tests-hs-src/Tests/HTTP1Parse.hs +46/−0
• tests/tests-hs-src/Tests/HTTP2Session.hs +584/−0
• tests/tests-hs-src/Tests/Utils.hs +64/−0
• tests/tests-hs-src/hunit_tests.hs +20/−15

@@ -41,27 +41,13 @@ Example ------- -There is a very basic example at `tests/tests-hs-src/compiling_ok.hs`.+There is a very basic example at `tests/tests-hs-src/compiling_ok.hs`, and a somewhat more complicated one at+`examples/attempt_bust`; that one shows how to do HTTP/2.0 push from the library. -Roadmap-------- -Done:--- Version 0.1: Having something that can run. No unit-testing, nothing-               fancy.--- Version 0.2: Absolutely minimal amount of unit tests.--- Version 0.3: More sensible logging.--Pending:--- Benchmarking.--Internal---------+Development+----------- -Uploading documentation (provided you have access to the package):+Uploading documentation (provided you have access to the package in Hackage):      $ ./hackage-upload-docs.sh second-transfer 0.5.4.0 <hackage-user> <hackage-password>

@@ -26,6 +26,50 @@ #include <openssl/ssl.h> #include <openssl/err.h> ++//+++// Some constants+#define ALL_OK  0+#define BAD_HAPPENED 1+#define TIMEOUT_REACHED 3+// This is also a failed IO with SSL, but this one may be quite+// natural and we want to handle it differently+#define TRANSPORT_CLOSED 2+++#if OPENSSL_VERSION_NUMBER < 0x1000200fL+#define ALPN_DISABLED+#warning Due to low OpenSSL version, this build is completely broken. Check second-transfer.cabal to see how+#warning to indicate a recent OpenSSL version.++int SSL_CTX_set_ecdh_auto(SSL *s, int onoff)+{+    fprintf(stderr, "Fake OPENSSL function called. Updated your SSL version!!\n");+}++void SSL_CTX_set_alpn_select_cb(SSL_CTX *ctx,+                                  int (*cb) (SSL *ssl,+                                             const unsigned char **out,+                                             unsigned char *outlen,+                                             const unsigned char *in,+                                             unsigned int inlen,+                                             void *arg),+                                void *arg)+{+    fprintf(stderr, "Fake OPENSSL function called. Updated your SSL version!!\n");+}+++void SSL_get0_alpn_selected(SSL*a , int*b,  int*c)+{+    fprintf(stderr, "Fake OPENSSL function called. Updated your SSL version!!\n");+}++#endif++ // Simple structure to keep track of the handle, and // of what needs to be freed later. typedef struct {@@ -49,12 +93,7 @@ // Call when you are done void close_connection(connection_t* conn); // Wait for the next one...-#define ALL_OK  0-#define BAD_HAPPENED 1-#define TIMEOUT_REACHED 3-// This is also a failed IO with SSL, but this one may be quite-// natural and we want to handle it differently-#define TRANSPORT_CLOSED 2+ int wait_for_connection(connection_t* conn, int microseconds, wired_session_t** wired_session); int send_data(wired_session_t* ws, char* buffer, int buffer_size); int recv_data(wired_session_t* ws, char* inbuffer, int buffer_size, int* data_recvd);

@@ -1,3 +1,8 @@+- 0.7.1.0 :+    * A few important memory leaks got fixed+    * Changed I/O interface to live in IOCallbacks+    * Working now around congestion window.+ - 0.6.1.0 :     * Improved documentation.     * Improved ALPN negotiation handling.

@@ -6,14 +6,14 @@ Stability   : experimental Portability : POSIX -SecondTransfer is a HTTP/1.1 and HTTP/2 server session library, with an emphasis towards+SecondTransfer is a HTTP\/1.1 and HTTP\/2 server session library, with an emphasis towards experimentation (so far). -This library implements enough of the HTTP/2  to build-compliant HTTP/2 servers. It also implements enough of-HTTP/1.1 so you can actually use it to build polyglot web-servers.+This library implements enough of the HTTP\/2  to build+compliant HTTP\/2 servers. It also implements enough of+HTTP\/1.1 so you can actually use it to build polyglot web-servers. -For HTTP/2, frame encoding and decoding is done with+For HTTP\/2, frame encoding and decoding is done with Kazu Yamamoto's <http://hackage.haskell.org/package/http2 http2> package. This library just takes care of making sense of sent and received frames.@@ -111,7 +111,8 @@ most times you can do with a simplified version called `CoherentWorker`. The function `coherentToAwareWorker` does the conversion. The difference between the two callbacks is  the level of information that you manage. With AwareWorker, you get a record on the request-with all sort of details, things like the session id and the +with all sort of details, things like the session id, the protocol the client is using and in+the future things like the remote address.   The callback is used to handle all requests to the server on a given negotiated ALPN protocol. If you need routing functionality (and you most certainly will need it), you need

@@ -9,7 +9,9 @@     ,BadPrefaceException (..)     ,HTTP11Exception (..)     ,HTTP11SyntaxException (..)+    ,ClientSessionAbortedException(..)     ,HTTP500PrecursorException (..)+    ,ConnectionCloseReason(..)     ,convertHTTP500PrecursorExceptionToException     ,getHTTP500PrecursorExceptionFromException     ,ContentLengthMissingException (..)@@ -55,6 +57,25 @@     toException   = convertHTTP2SessionExceptionToException     fromException = getHTTP2SessionExceptionFromException +-- TODO below: add the other protocol reasons++-- | Reasons for a remote server interrupting a connectionn of this client+data ConnectionCloseReason =+    NormalTermination_CCR     -- ^ Corresponds to NO_ERROR+    |SessionAlreadyClosed_CCR -- ^ A request was done after the session was previously closed.+    |IOChannelClosed_CCR      -- ^ This one happens when one of the IO channels is closed and a BlockedIndefinitelyOnMVar bubbles up. It should only happen in the test suite, as the OpenSSL_TLS channel uses a specialized exception type. If you see it in the wild, it is a bug.+    |ProtocolError_CCR        -- ^ Any other reason+    deriving Show++-- | Concrete Exception. Used internally to signal that the server broke+--   the connection. This is a public exception that clients of the library+--   will see when acting as an HTTP client.+data ClientSessionAbortedException = ClientSessionAbortedException ConnectionCloseReason+    deriving (Typeable, Show)++instance Exception ClientSessionAbortedException where+    toException = convertHTTP2SessionExceptionToException+    fromException = getHTTP2SessionExceptionFromException  -- | Abstract exception. Thrown when encoding/decoding of a frame fails data FramerException = forall e . Exception e => FramerException e

@@ -48,10 +48,10 @@         forkIO $ go new_session_tag (Just "") 1         return ()   where-    push_action = attendant_callbacks ^. pushAction_AtC-    -- pull_action = attendant_callbacks ^. pullAction_AtC-    close_action = attendant_callbacks ^. closeAction_AtC-    best_effort_pull_action = attendant_callbacks ^. bestEffortPullAction_AtC+    push_action = attendant_callbacks ^. pushAction_IOC+    -- pull_action = attendant_callbacks ^. pullAction_IOC+    close_action = attendant_callbacks ^. closeAction_IOC+    best_effort_pull_action = attendant_callbacks ^. bestEffortPullAction_IOC      go :: Int -> Maybe B.ByteString -> Int -> IO ()     go session_tag (Just leftovers) reuse_no = do

@@ -1,3 +1,4 @@+ -- 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,@@ -5,6 +6,7 @@ {-# OPTIONS_HADDOCK hide #-} module SecondTransfer.Http2.Framer (     BadPrefaceException,+    SessionPayload(..),      wrapSession,     http2FrameLength,@@ -46,6 +48,7 @@  import qualified Data.HashTable.IO                      as H import           System.Clock                           (Clock(..),getTime)+-- import           System.Mem.Weak  import           SecondTransfer.Sessions.Internal       (                                                          sessionExceptionHandler,@@ -119,19 +122,34 @@   data FramerSessionData = FramerSessionData {-      _stream2flow           :: MVar Stream2AvailSpace++    -- A dictionary (protected by a lock) from stream id to flow control command.+    _stream2flow           :: MVar Stream2AvailSpace+     -- Two members below: the place where you put data which is going to be flow-controlled,     -- and the place with the ordinal+    -- Flow control dictionary. It goes from stream id to the stream data-output gate (an-mvar)+    -- and to a mutable register for the ordinal. The ordinal for packets inside a stream is used+    -- for priority and reporting     , _stream2outputBytes    :: MVar ( HashTable GlobalStreamId (MVar LB.ByteString, MVar Int) )++    -- The default flow-control window advertised by the peer (e.g., the browser)     , _defaultStreamWindow   :: MVar Int -    -- Wait variable to output bytes to the channel+    -- Wait variable to output bytes to the channel. This is needed to avoid output data races.     , _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++    -- Exclusive lock for headers, needed since the specs forbid interleaving header and data     , _noHeadersInChannel    :: MVar NoHeadersInChannel++    -- The push action     , _pushAction            :: PushAction++    -- The close action     , _closeAction           :: CloseAction      -- Global id of the session, used for e.g. error reporting.@@ -141,10 +159,19 @@     , _sessionsContext       :: SessionsContext      -- For GoAway frames-    , _lastStream            :: MVar Int+    -- We keep the value below updated with the highest incoming stream+    -- frame. For example, it is updated as soon as headers are received from+    -- the client on that stream.+    , _lastInputStream       :: MVar Int+    -- We update this one as soon as an outgoing frame is seen with such a+    -- high output number.+    , _lastOutputStream      :: MVar Int      -- For sending data orderly     , _prioritySendState     :: PrioritySendState++    -- Need to know this for the preface+    , _sessionRole_FSD       :: SessionRole     }  L.makeLenses ''FramerSessionData@@ -152,27 +179,41 @@  type FramerSession = ReaderT FramerSessionData IO +data SessionPayload =+    AwareWorker_SP AwareWorker   -- I'm a server+    |ClientState_SP ClientState  -- I'm a client -wrapSession :: AwareWorker -> SessionsContext -> Attendant-wrapSession aware_worker sessions_context attendant_callbacks = do+wrapSession :: SessionPayload -> SessionsContext -> Attendant+wrapSession session_payload sessions_context io_callbacks = do      let         session_id_mvar = view nextSessionId sessions_context-        push_action = attendant_callbacks ^. pushAction_AtC-        pull_action = attendant_callbacks ^. pullAction_AtC-        close_action = attendant_callbacks ^. closeAction_AtC-        best_effort_pull_action = attendant_callbacks ^. bestEffortPullAction_AtC+        push_action = io_callbacks ^. pushAction_IOC+        pull_action = io_callbacks ^. pullAction_IOC+        close_action = io_callbacks ^. closeAction_IOC+        -- best_effort_pull_action = io_callbacks ^. bestEffortPullAction_IOC       new_session_id <- modifyMVarMasked         session_id_mvar $         \ session_id -> return $ session_id `seq` (session_id + 1, session_id) -    (session_input, session_output) <- http2Session-                                        aware_worker-                                        new_session_id-                                        sessions_context+    (session_input, session_output) <- case session_payload of+        AwareWorker_SP aware_worker ->  http2ServerSession+                                            aware_worker+                                            new_session_id+                                            sessions_context+        ClientState_SP client_state -> http2ClientSession+                                            client_state+                                            new_session_id+                                            sessions_context +    let+        session_role = case session_payload of+            AwareWorker_SP _ ->  Server_SR+            ClientState_SP _ ->  Client_SR++     -- TODO : Add type annotations....     s2f                       <- H.new     stream2flow_mvar          <- newMVar s2f@@ -182,6 +223,7 @@     can_output                <- newMVar CanOutput     no_headers_in_channel     <- newMVar NoHeadersInChannel     last_stream_id            <- newMVar 0+    last_output_stream_id     <- newMVar 0     output_is_forbidden       <- newMVar False      prio_mvar                 <- STM.atomically $ newTMVar PQ.empty@@ -200,12 +242,14 @@         ,_closeAction         = close_action         ,_sessionIdAtFramer   = new_session_id         ,_sessionsContext     = sessions_context-        ,_lastStream          = last_stream_id+        ,_lastInputStream     = last_stream_id+        ,_lastOutputStream    = last_output_stream_id         ,_outputIsForbidden   = output_is_forbidden         ,_prioritySendState   = PrioritySendState {                                     _semToSend = sem_to_send,                                     _prioQ = prio_mvar                                 }+        ,_sessionRole_FSD     = session_role         }  @@ -226,7 +270,7 @@         exc_handler :: Int -> SessionsContext -> FramerException -> IO ()         exc_handler x y e = do             modifyMVar_ output_is_forbidden (\ _ -> return True)-            INSTRUMENTATION( errorM "HTTP2.Framer" "Exception went up" )+            -- INSTRUMENTATION( errorM "HTTP2.Framer" "Exception went up" )             sessionExceptionHandler Framer_HTTP2SessionComponent x y e          io_exc_handler :: Int -> SessionsContext -> IOProblem -> IO ()@@ -265,6 +309,10 @@         GlobalStreamId ->         Int           ->         FramerSession Bool+addCapacity _         0         =+    -- By the specs, a WINDOW_UPDATE with 0 of credit should be considered a protocol+    -- error+    return False addCapacity 0         delta_cap =     -- TODO: Implement session flow control     return True@@ -273,7 +321,7 @@         table_mvar <- view stream2flow         val <- liftIO $ withMVar table_mvar $ \ table ->             H.lookup table stream_id-        last_stream_mvar <- view lastStream+        last_stream_mvar <- view lastInputStream         last_stream <- liftIO . readMVar $ last_stream_mvar         case val of             Nothing | stream_id > last_stream ->@@ -340,25 +388,30 @@ -- -- This function also does part of the flow control: it registers WindowUpdate frames and triggers -- quota updates on the streams.+--+-- Also, when this function exits its caller will issue a close_action inputGatherer :: PullAction -> SessionInput -> FramerSession () inputGatherer pull_action session_input = do-    -- We can start by reading off the prefix....-    prefix <- liftIO $ pull_action http2PrefixLength-    if prefix /= NH2.connectionPreface-      then do-        sendGoAwayFrame NH2.ProtocolError-        liftIO $-            -- We just use the GoAway frame, although this is awfully early-            -- and probably wrong-            throwIO BadPrefaceException-      else-        INSTRUMENTATION(  debugM "HTTP2.Framer" "Prologue validated" )++    session_role <- view sessionRole_FSD++    when (session_role == Server_SR) $ do+        -- We can start by reading off the prefix....+        prefix <- liftIO $ pull_action http2PrefixLength+        when (prefix /= NH2.connectionPreface) $ do+            sendGoAwayFrame NH2.ProtocolError+            liftIO $+                -- We just use the GoAway frame, although this is awfully early+                -- and probably wrong+                throwIO BadPrefaceException+     let         source::Source FramerSession (Maybe NH2.Frame)         source = transPipe liftIO $ readNextFrame pull_action     source $$ consume True   where +     sendToSession :: Bool -> InputFrame -> IO ()     sendToSession starting frame =       -- print(NH2.streamId $ NH2.frameHeader frame)@@ -383,9 +436,15 @@     consume starting = do         maybe_maybe_frame <- await -        case maybe_maybe_frame of+        output_is_forbidden_mvar <- view outputIsForbidden+        output_is_forbidden <- liftIO $ readMVar output_is_forbidden_mvar +        -- Consumption ends automatically when the output is forbidden, this might help avoiding+        -- attacks where a peer refuses to close its socket.+        unless output_is_forbidden $ case maybe_maybe_frame of+             Just Nothing      ->+                -- Only way to get here is by a closed connection condition I guess.                 abortSession              Just (Just right_frame) -> do@@ -395,7 +454,10 @@                         -- Bookkeep the increase on bytes on that stream                         -- liftIO $ putStrLn $ "Extra capacity for stream " ++ (show stream_id)                         succeeded <- lift $ addCapacity stream_id (fromIntegral credit)-                        unless succeeded abortSession+                        if not succeeded then+                            abortSession+                        else+                            consume_continue                       frame@(NH2.Frame _ (NH2.SettingsFrame settings_list) ) -> do@@ -426,18 +488,18 @@                         -- And send the frame down to the session, so that session specific settings                         -- can be applied.                         liftIO $ sendToSession starting $! frame-+                        consume_continue                      a_frame@(NH2.Frame (NH2.FrameHeader _ _ stream_id) _ )   -> do                         -- Update the keep of last stream                         -- lift . startStreamOutputQueueIfNotExists (NH2.fromStreamIdentifier stream_id) priority-                        lift . updateLastStream $ stream_id+                        lift . updateLastInputStream $ stream_id                          -- Send frame to the session                         liftIO $ sendToSession starting a_frame -                -- tail recursion: go again...-                consume_continue+                        -- tail recursion: go again...+                        consume_continue              Nothing    ->                 -- We may as well exit this thread@@ -447,6 +509,13 @@ -- All the output frames come this way first outputGatherer :: SessionOutput -> FramerSession () outputGatherer session_output = do++    session_role <- view sessionRole_FSD++    -- When acting as a client, the first step is to send the prefix...+    when (session_role == Client_SR) $+        pushPrefix+     frame_sent_report_callback <- view $        sessionsContext            .        sessionsConfig             .@@ -454,7 +523,6 @@        dataDeliveryCallback_SC      session_id <- view sessionIdAtFramer-     let         dataForFrame p1 p2 =@@ -467,16 +535,24 @@            command_or_frame  <- liftIO $ getFrameFromSession session_output            case command_or_frame of -               Left CancelSession_SOC -> do-                   -- The session wants to cancel things-                   INSTRUMENTATION(  debugM "HTTP2.Framer" "CancelSession_SOC processed")+               Left (CancelSession_SOC error_code) -> do+                   -- The session wants to cancel things as harshly as possible, send a GoAway frame with+                   -- the information I have here.+                   sendGoAwayFrame error_code                    releaseFramer+                   -- And this causes this thread to finish, so that no new frames+                   -- are taken from the session. Correspondingly, an exception is raised in+                   -- the session if it tries to write another frame -               Left (FinishStream_SOC stream_id ) ->+               Left (SpecificTerminate_SOC last_stream_id) -> do+                   -- This is used when the session wants to finish in a specific way.+                   sendSpecificTerminateGoAway last_stream_id+                   releaseFramer++               Left (FinishStream_SOC stream_id ) -> do                    -- Session knows that we are done with the given stream, and that we can release                    -- the flow control structures-                   -- NOTICE: Unfortunately, this doesn't work, so ignore the message for now-                   -- finishFlowControlForStream stream_id.+                   finishFlowControlForStream stream_id                    cont                 Right ( p1@(NH2.EncodeInfo _ stream_idii _), p2@(NH2.DataFrame _), ef ) -> do@@ -546,11 +622,18 @@     loopPart session_id frame_sent_report_callback  -updateLastStream :: GlobalStreamId  -> FramerSession ()-updateLastStream stream_id = do-    last_stream_id_mvar <- view lastStream+updateLastInputStream :: GlobalStreamId  -> FramerSession ()+updateLastInputStream stream_id = do+    last_stream_id_mvar <- view lastInputStream     liftIO $ modifyMVar_ last_stream_id_mvar (\ x -> return $ max x stream_id) ++updateLastOutputStream :: GlobalStreamId  -> FramerSession ()+updateLastOutputStream stream_id = do+    last_stream_id_mvar <- view lastOutputStream+    liftIO $ modifyMVar_ last_stream_id_mvar (\ x -> return $ max x stream_id)++ startStreamOutputQueueIfNotExists :: GlobalStreamId -> Int -> FramerSession () startStreamOutputQueueIfNotExists stream_id priority = do     table_mvar <- view stream2flow@@ -574,13 +657,12 @@      s2o_mvar <- view stream2outputBytes -    liftIO . withMVar s2o_mvar $ \ s2o ->  H.insert s2o stream_id (bytes_chan, ordinal_num)+    liftIO . withMVar s2o_mvar $ {-# SCC e1  #-}  \ s2o ->  H.insert s2o stream_id (bytes_chan, ordinal_num)      stream2flow_mvar <- view stream2flow  -    liftIO . withMVar stream2flow_mvar  $ \ s2c ->  H.insert s2c stream_id command_chan-+    liftIO . withMVar stream2flow_mvar  $  {-# SCC e2  #-}  \ s2c ->  H.insert s2c stream_id command_chan     --     initial_cap_mvar <- view defaultStreamWindow     initial_cap <- liftIO $ readMVar initial_cap_mvar@@ -592,23 +674,22 @@     -- 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+            close_action       read_state <- ask     liftIO $ forkIO $ close_on_error session_id' sessions_context  $ runReaderT-        (flowControlOutput stream_id priority initial_cap 0 "" command_chan bytes_chan)+        ({-# SCC forkFlowControlOutput  #-} flowControlOutput stream_id priority initial_cap 0 "" command_chan bytes_chan)         read_state      return (bytes_chan , command_chan)@@ -670,15 +751,31 @@     sendBytes bs  +pushPrefix :: FramerSession ()+pushPrefix = do+    let bs = LB.fromStrict NH2.connectionPreface+    sendBytes bs+++-- Default sendGoAwayFrame. This one assumes that actions are taken as soon as stream is+-- received. This is a good default strategy for streams that cause things to happen, that+-- is, the ones with POST and GET. sendGoAwayFrame :: NH2.ErrorCodeId -> FramerSession () sendGoAwayFrame error_code = do-    last_stream_id_mvar <- view lastStream+    last_stream_id_mvar <- view lastInputStream     last_stream_id <- liftIO $ readMVar last_stream_id_mvar     pushFrame         (NH2.EncodeInfo NH2.defaultFlags 0 Nothing)         (NH2.GoAwayFrame last_stream_id error_code "")  +sendSpecificTerminateGoAway :: GlobalStreamId -> FramerSession ()+sendSpecificTerminateGoAway last_stream =+    pushFrame+        (NH2.EncodeInfo NH2.defaultFlags 0 Nothing)+        (NH2.GoAwayFrame last_stream NH2.NoError "")++ -- From this point on data is really serialized, sendBytes :: LB.ByteString -> FramerSession () sendBytes bs = do@@ -699,14 +796,21 @@ -- -- There is one of these for each stream ---flowControlOutput :: Int -> Int -> Int -> Int ->  LB.ByteString -> MVar FlowControlCommand -> MVar LB.ByteString ->  FramerSession ()+flowControlOutput :: Int+                     -> Int+                     -> Int+                     -> Int+                     ->  LB.ByteString+                     -> MVar FlowControlCommand+                     -> MVar LB.ByteString+                     ->  FramerSession () flowControlOutput stream_id priority capacity ordinal leftovers commands_chan bytes_chan =-    if leftovers == ""-      then do+    ordinal `seq` if leftovers == ""+      then {-# SCC fcOBranch1  #-} do         -- Get more data (possibly block waiting for it)-        bytes_to_send <- liftIO $ takeMVar bytes_chan+        bytes_to_send <- liftIO $ {-# SCC perfectlyHarmlessExceptionPoint #-} takeMVar bytes_chan         flowControlOutput stream_id priority capacity ordinal  bytes_to_send commands_chan bytes_chan-      else do+      else {-# SCC fcOBranch2  #-}  do         -- Length?         let amount = fromIntegral  (LB.length leftovers - 9)         if  amount <= capacity@@ -715,11 +819,11 @@             -- I can send ... if no headers are in process....             -- liftIO . logit $ "set-priority (stream_id, prio, ordinal) " `mappend` (pack . show) (__stream_id, priority, ordinal)             withPrioritySend priority stream_id ordinal leftovers-            flowControlOutput  stream_id priority (capacity - amount) (ordinal+1) "" commands_chan bytes_chan+            flowControlOutput  stream_id priority (capacity - amount) (ordinal+1 ) "" commands_chan bytes_chan           else do             -- I can not send because flow-control is full, wait for a command instead-            command <- liftIO $ takeMVar commands_chan-            case command of+            command <- liftIO $ {-# SCC t2 #-} takeMVar commands_chan+            case  {-# SCC t3 #-} command of                 AddBytes_FCM delta_cap ->                     -- liftIO $ putStrLn $ "Flow control delta_cap stream " ++ (show stream_id)                     flowControlOutput stream_id priority (capacity + delta_cap) ordinal leftovers commands_chan bytes_chan@@ -728,7 +832,8 @@ releaseFramer :: FramerSession () releaseFramer =     -- Release any resources pending...-+    -- This is not needed as of right now, since garbage collection works well.+    -- I'm leaving it here just in case we need to re-activate it in the future.     return ()  

@@ -4,24 +4,20 @@     ) where  -import           SecondTransfer.Http2.Framer            (wrapSession)+import           SecondTransfer.Http2.Framer            (wrapSession, SessionPayload(..)) import           SecondTransfer.Sessions.Internal       (SessionsContext) import           SecondTransfer.MainLoop.CoherentWorker import           SecondTransfer.MainLoop.PushPullType   (Attendant) --- | The type of this function is equivalent to:------ @---      http2Attendant :: CoherentWorker -> AttendantCallbacks ->  IO ()--- @+-- | ----- Given a `CoherentWorker`, this function wraps it with flow control, multiplexing,+-- Given an `AwareWorker`, this function wraps it with flow control, multiplexing, -- and state maintenance needed to run an HTTP/2 session. ----- Notice that this function is  using HTTP/2 over TLS. We haven't implemented yet--- a session handling mechanism for HTTP/1.1 .+-- Notice that this function is  using HTTP/2 over TLS. There is an equivalent for HTTP+-- 1.1. http2Attendant :: SessionsContext -> AwareWorker -> Attendant http2Attendant sessions_context coherent_worker attendant_callbacks = do     let-        attendant = wrapSession coherent_worker sessions_context+        attendant = wrapSession (AwareWorker_SP coherent_worker) sessions_context     attendant attendant_callbacks

@@ -4,11 +4,14 @@ {-# LANGUAGE FlexibleContexts, Rank2Types, TemplateHaskell, OverloadedStrings #-} {-# OPTIONS_HADDOCK hide #-} module SecondTransfer.Http2.Session(-    http2Session+    http2ServerSession+    ,http2ClientSession     ,getFrameFromSession     ,sendFirstFrameToSession     ,sendMiddleFrameToSession     ,sendCommandToSession+    ,makeClientState+    ,pendingRequests_ClS      ,CoherentSession     ,SessionInput(..)@@ -20,6 +23,8 @@     ,SessionsCallbacks(..)     ,SessionsConfig(..)     ,ErrorCallback+    ,ClientState(..)+    ,SessionRole(..)      -- Internal stuff     ,OutputFrame@@ -29,7 +34,7 @@ #include "Logging.cpphs"  -- System grade utilities-import           Control.Concurrent                     (ThreadId, forkIO, threadDelay)+import           Control.Concurrent                     (ThreadId, forkIO) import           Control.Concurrent.Chan import           Control.Exception                      (throwTo) import qualified Control.Exception                      as E@@ -48,6 +53,7 @@ import qualified Data.HashTable.IO                      as H import qualified Data.IntSet                            as NS import           Data.Maybe                             (isJust)+import           Data.Typeable #ifndef IMPLICIT_MONOID import           Data.Monoid                            (mappend) #endif@@ -67,6 +73,8 @@ import           SecondTransfer.MainLoop.CoherentWorker import           SecondTransfer.MainLoop.Tokens import           SecondTransfer.MainLoop.Protocol+import           SecondTransfer.MainLoop.ClientPetitioner+ import           SecondTransfer.Sessions.Config import           SecondTransfer.Sessions.Internal       (sessionExceptionHandler,                                                          SessionsContext,@@ -74,8 +82,10 @@ import           SecondTransfer.Utils                   (unfoldChannelAndSource) import           SecondTransfer.Exception import qualified SecondTransfer.Utils.HTTPHeaders       as He-import           SecondTransfer.MainLoop.Logging        (logWithExclusivity, logit)+import           SecondTransfer.MainLoop.Logging        (logit) +--import           Debug.Trace+ -- Unfortunately the frame encoding API of Network.HTTP2 is a bit difficult to -- use :-( type OutputFrame = (NH2.EncodeInfo, NH2.FramePayload, Effect)@@ -93,11 +103,16 @@ -- end of data. Middle value is delay in microseconds type DataOutputToConveyor = (GlobalStreamId, Maybe B.ByteString, Effect) -+-- What to do regarding headers data HeaderOutputMessage =+    -- Send the headers of the principal stream     NormalResponse_HM (GlobalStreamId, MVar HeadersSent, Headers, Effect)-    --+    -- Send a push-promise     |PushPromise_HM   (GlobalStreamId, GlobalStreamId, Headers, Effect)+    -- Send a reset stream notification for the stream given below+    |ResetStream_HM   (GlobalStreamId, Effect)+    -- Send a GoAway, where last-stream is the stream given below.+    |GoAway_HM (GlobalStreamId, Effect)   -- Settings imposed by the peer@@ -179,13 +194,27 @@   deriving Show  --- temporary+-- Session output commands indicating something to the Framer data  SessionOutputCommand =-    CancelSession_SOC+    -- Command sent to the framer to close the session as harshly as possible. The framer+    -- will pick its own last valid stream.+    CancelSession_SOC NH2.ErrorCodeId+    -- Command sent to the framer to close the session specifying a specific last-stream.+    -- The reason used will be "NoError"+    |SpecificTerminate_SOC GlobalStreamId+    -- Command sent to the framer to notify it of a normal end of stream+    -- condition, this is used to naturally close each stream.     |FinishStream_SOC GlobalStreamId   deriving Show  +-- The role of a session is either server or client. There are small+-- differences between both.+data SessionRole  =+    Client_SR+   |Server_SR+  deriving (Eq,Show)+ -- Here is how we make a session type SessionMaker = SessionsContext -> IO Session @@ -195,7 +224,78 @@  data PostInputMechanism = PostInputMechanism (MVar (Maybe B.ByteString), InputDataStream) +------------- Regarding client state+type Message = (Headers,InputDataStream) +type RequestResult = Either ConnectionCloseReason Message++data ClientState = ClientState {+    -- Holds a queue. The client here puts the message (the request) and an VAR+    -- where it will receive the response.+    _pendingRequests_ClS       :: MVar (Message, MVar RequestResult)++    -- This is the id of the next available stream+    ,_nextStream_ClS           :: MVar Int++    -- Says if the client has been closed+    ,_clientIsClosed_ClS      :: MVar Bool++    -- A dictionary from stream id to the client which is waiting for the+    -- message+    ,_response2Waiter_ClS      ::HashTable GlobalStreamId (MVar RequestResult)+    }++makeLenses ''ClientState+++makeClientState :: IO ClientState+makeClientState = do+    request_chan <- newEmptyMVar+    next_stream_mvar <- newMVar 3+    client_is_closed_mvar <- newMVar False+    new_h <- H.new++    return ClientState {+        _pendingRequests_ClS = request_chan+        ,_nextStream_ClS = next_stream_mvar+        ,_response2Waiter_ClS = new_h+        ,_clientIsClosed_ClS = client_is_closed_mvar+        }+++handleRequest' :: ClientState -> Headers -> InputDataStream -> IO (Headers,InputDataStream)+handleRequest' client_state headers input_data = runReaderT (handleRequest headers input_data) client_state++type ClientMonad = ReaderT ClientState IO++handleRequest :: Headers -> InputDataStream -> ClientMonad Message+handleRequest headers input_data = do+    pending_requests <- view pendingRequests_ClS+    response_mvar <- liftIO $ newEmptyMVar++    liftIO $ E.catch+       (do+           {-# SCC cause1 #-} putMVar pending_requests ((headers,input_data),response_mvar)+           either_reason_or_message <- {-# SCC cause2 #-} liftIO $ takeMVar response_mvar+           case either_reason_or_message of+               Left break_reason -> E.throw $ ClientSessionAbortedException break_reason+               Right message -> return message+       )+       ( ( \ _ -> E.throw $ ClientSessionAbortedException SessionAlreadyClosed_CCR ):: E.BlockedIndefinitelyOnMVar -> IO Message )+++++instance ClientPetitioner ClientState where++    request = handleRequest'++-------------- Regarding client state+++-- SessionData is the actual state of the session, including the channels to the framer+-- outside.+-- -- NH2.Frame != Frame data SessionData = SessionData {     -- ATTENTION: Ignore the warning coming from here for now@@ -209,12 +309,15 @@      -- 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@@ -229,8 +332,13 @@     -- I make copies of it in different contexts, and as needed.     ,_forWorkerThread            :: WorkerThreadEnvironment +    -- When acting as a server, this is the handler for processing requests     ,_awareWorker                :: AwareWorker +    -- When acting as a client, this is where new requests are taken+    -- from+    ,_simpleClient               :: ClientState+     -- Some streams may be cancelled     ,_streamsCancelled           :: MVar NS.IntSet @@ -251,15 +359,28 @@      -- What is the next stream available for push?     ,_nextPushStream             :: MVar Int++    -- What role does this session has?+    ,_sessionRole                :: SessionRole+     }   makeLenses ''SessionData  +http2ServerSession :: AwareWorker -> Int -> SessionsContext -> IO Session+http2ServerSession a i sctx = http2Session Server_SR a (error "NotAClient") i sctx+++http2ClientSession :: ClientState -> Int -> SessionsContext -> IO Session+http2ClientSession client_state session_id sctx =+    http2Session Client_SR (error "NotAServer") client_state session_id sctx++ --                                v- {headers table size comes here!!}-http2Session :: AwareWorker -> Int -> SessionsContext -> IO Session-http2Session aware_worker session_id sessions_context =   do+http2Session :: SessionRole -> AwareWorker -> ClientState  -> Int -> SessionsContext -> IO Session+http2Session session_role aware_worker client_state session_id sessions_context =   do     session_input             <- newChan     session_output            <- newChan     session_output_mvar       <- newMVar session_output@@ -309,6 +430,7 @@         ,_stream2HeaderBlockFragment = stream_request_headers         ,_forWorkerThread            = for_worker_thread         ,_awareWorker                = aware_worker+        ,_simpleClient               = client_state         ,_streamsCancelled           = cancelled_streams_mvar         ,_stream2PostInputMechanism  = stream2postinputmechanism         ,_stream2WorkerThread        = stream2workerthread@@ -317,18 +439,60 @@         ,_sessionSettings            = session_settings         ,_lastGoodStream             = last_good_stream_mvar         ,_nextPushStream             = next_push_stream+        ,_sessionRole                = session_role         }      let+        io_exc_handler :: SessionComponent -> E.BlockedIndefinitelyOnMVar -> IO ()+        io_exc_handler _component _e = do+           case session_role of+               Server_SR -> do+                   -- sessionExceptionHandler component session_id sessions_context e+                   -- Just ignore this kind of exceptions here, very explicitly, as they should naturally+                   -- happen when the framer is closed+                   return ()++               Client_SR -> do+                   clientSideTerminate client_state IOChannelClosed_CCR++         exc_handler :: SessionComponent -> HTTP2SessionException -> IO ()-        exc_handler component e = sessionExceptionHandler component session_id sessions_context e+        exc_handler component e = do+            case session_role of+                Server_SR ->+                    sessionExceptionHandler component session_id sessions_context e++                Client_SR -> do+                    let+                        maybe_client_session_aborted :: Maybe ClientSessionAbortedException+                        maybe_client_session_aborted | HTTP2SessionException ee <- e  = cast ee++                        maybe_protocol_error :: Maybe HTTP2ProtocolException+                        maybe_protocol_error | HTTP2SessionException ee <- e = cast ee++                    case maybe_client_session_aborted of+                        Just (ClientSessionAbortedException reason) -> do+                            clientSideTerminate client_state reason+++                        Nothing ->+                            if isJust maybe_protocol_error+                                then+                                    clientSideTerminate client_state ProtocolError_CCR+                                else do+                                    E.throw 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-            )+        exc_guard component action =+                E.catch+                    (E.catch+                         action+                         (\e -> do+                             -- INSTRUMENTATION( errorM "HTTP2.Session" "Exception processed" )+                             exc_handler component e+                         )+                    )+                    (io_exc_handler component)          use_chunk_length =  sessions_context ^.  sessionsConfig . dataFrameSize @@ -345,6 +509,13 @@     forkIO $ exc_guard SessionDataOutputThread_HTTP2SessionComponent            $ dataOutputThread use_chunk_length data_output session_output_mvar +    -- If I'm a client, I also need a thread to poll for requests+    when (session_role == Client_SR) $ do+         forkIO $ exc_guard SessionClientPollThread_HTTP2SessionComponent+                $ sessionPollThread session_data headers_output+         return ()++     -- 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.@@ -362,18 +533,19 @@     -- these values in a closure...     session_input             <- view sessionInput -    decode_headers_table_mvar <- view toDecodeHeaders-    stream_request_headers    <- view stream2HeaderBlockFragment+    -- decode_headers_table_mvar <- view toDecodeHeaders+    -- stream_request_headers    <- view stream2HeaderBlockFragment     cancelled_streams_mvar    <- view streamsCancelled-    coherent_worker           <- view awareWorker+    -- coherent_worker           <- view awareWorker -    for_worker_thread_uns     <- view forWorkerThread+    -- for_worker_thread_uns     <- view forWorkerThread     stream2workerthread       <- view stream2WorkerThread-    receiving_headers_mvar    <- view receivingHeaders-    last_good_stream_mvar     <- view lastGoodStream-    current_session_id        <- view sessionIdAtSession+    -- receiving_headers_mvar    <- view receivingHeaders+    -- last_good_stream_mvar     <- view lastGoodStream+    -- current_session_id        <- view sessionIdAtSession -    input                     <- liftIO $ readChan session_input+    input                     <- {-# SCC session_input #-} liftIO $ readChan session_input+    session_role              <- view sessionRole      case input of @@ -423,145 +595,19 @@         -- 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.-        MiddleFrame_SIC 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-                    -- 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( errorM "HTTP2.Session" "Protocol error: bad stream id")-                        closeConnectionBecauseIsInvalid NH2.ProtocolError-                -- So, now we know, we are ignoring priority frames at the moment-                -- liftIO . logit $ "Priority: " `mappend` (pack . show $ getHeadersPriority frame)-              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 )-                -- Lets get a time-                headers_arrived_time      <- liftIO $ getTime Monotonic-                -- 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-#if LOGIT_SWITCH_TIMINGS-                let-                    (Just path) = He.fetchHeader header_list ":path"-                liftIO $ logit $ (pack . show $ stream_id ) `mappend` " -> " `mappend` path-#endif-                -- /DEBUG-                -- 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--                let-                  perception = Perception {-                      _startedTime_Pr = headers_arrived_time,-                      _streamId_Pr = stream_id,-                      _sessionId_Pr = current_session_id,-                      _protocol_Pr = Http2_HPV,-                      _anouncedProtocols_Pr = Nothing-                      }-                  request = Request {-                      _headers_RQ = header_list_after,-                      _inputData_RQ = post_data_source,-                      _perception_RQ = perception-                      }--                -- 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-                ---                -- NOTE: Some late internal errors from the worker thread are-                --       handled here by clossing the session.-                ---                -- TODO: Log exceptions handled here.-                liftIO $ do-                    thread_id <- forkIO $ E.catch-                        (runReaderT-                            (workerThread-                                   request-                                   coherent_worker)-                            for_worker_thread-                        )-                        (-                            (   \ _ ->  do-                                -- Actions to take when the thread breaks....-                                writeChan session_input InternalAbort_SIC-                            )-                            :: HTTP500PrecursorException -> IO ()-                        )--                    H.insert stream2workerthread stream_id thread_id--                return ()-            else-                -- Frame doesn't end the headers... it was added before... so-                -- probably do nothing-                return ()+        MiddleFrame_SIC frame | Just (_stream_id, _bytes) <- isAboutHeaders frame ->+            case session_role of+                Server_SR -> do+                    -- Just append the frames to streamRequestHeaders+                    serverProcessIncomingHeaders frame+                    continue -            continue+                Client_SR -> do+                    clientProcessIncomingHeaders frame+                    continue +        -- Peer can order a stream aborted, meaning that we shall not send any more data+        -- on it.         MiddleFrame_SIC frame@(NH2.Frame _ (NH2.RSTStreamFrame _error_code_id)) -> do             let stream_id = streamIdFromFrame frame             liftIO $ do@@ -659,6 +705,19 @@             handleSettingsFrame settings_list             continue +        MiddleFrame_SIC (NH2.Frame _ (NH2.GoAwayFrame _ _ _ ))+            | Server_SR <- session_role -> do+                -- I was sent a go away, so go-away...+                closeConnectionBecauseIsInvalid NH2.NoError+                return ()++            | Client_SR <- session_role -> do+                -- I was sent a go away, so go-away, but use the kind of exception+                -- that will unwind the stack gracefully+                closeConnectionForClient NH2.NoError+                return ()++         MiddleFrame_SIC somethingelse ->  unlessReceivingHeaders $ do             -- An undhandled case here....             -- liftIO . logit $ "Strange frame:" `mappend` (pack . show $ somethingelse)@@ -691,6 +750,257 @@             (NH2.SettingsFrame [])  +serverProcessIncomingHeaders :: NH2.Frame ->  ReaderT SessionData IO ()+serverProcessIncomingHeaders frame | Just (stream_id, bytes) <- isAboutHeaders frame = do++    -- Just append the frames to streamRequestHeaders+    opens_stream              <- appendHeaderFragmentBlock stream_id bytes+    receiving_headers_mvar    <- view receivingHeaders+    last_good_stream_mvar     <- view lastGoodStream+    for_worker_thread_uns     <- view forWorkerThread+    decode_headers_table_mvar <- view toDecodeHeaders+    stream_request_headers    <- view stream2HeaderBlockFragment+    coherent_worker           <- view awareWorker+    current_session_id        <- view sessionIdAtSession+    session_input             <- view sessionInput+    stream2workerthread       <- view stream2WorkerThread++    if opens_stream+      then do+        maybe_rcv_headers_of <- liftIO $ takeMVar receiving_headers_mvar+        case maybe_rcv_headers_of of+          Just _ -> do+            -- Bad peer, it is already sending headers+            -- and trying to open another one+            {-# SCC ccB1 #-} 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( errorM "HTTP2.Session" "Protocol error: bad stream id")+                {-# SCC ccB2 #-} closeConnectionBecauseIsInvalid NH2.ProtocolError+        -- So, now we know, we are ignoring priority frames at the moment+        -- liftIO . logit $ "Priority: " `mappend` (pack . show $ getHeadersPriority frame)+      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 )+        -- Lets get a time+        headers_arrived_time      <- liftIO $ getTime Monotonic+        -- 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+#if LOGIT_SWITCH_TIMINGS+        let+            (Just path) = He.fetchHeader header_list ":path"+        liftIO $ logit $ (pack . show $ stream_id ) `mappend` " -> " `mappend` path+#endif+        -- /DEBUG+        -- 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 <- validateIncomingHeadersServer headers_editor++        good_headers <- case maybe_good_headers_editor of+            Just yes_they_are_good -> return yes_they_are_good+            Nothing -> {-# SCC ccB3 #-} 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++        let+          perception = Perception {+              _startedTime_Pr = headers_arrived_time,+              _streamId_Pr = stream_id,+              _sessionId_Pr = current_session_id,+              _protocol_Pr = Http2_HPV,+              _anouncedProtocols_Pr = Nothing+              }+          request = Request {+              _headers_RQ = header_list_after,+              _inputData_RQ = post_data_source,+              _perception_RQ = perception+              }++        -- 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+        --+        -- NOTE: Some late internal errors from the worker thread are+        --       handled here by clossing the session.+        --+        -- TODO: Log exceptions handled here.+        liftIO $ do+            thread_id <- forkIO $ E.catch+                (runReaderT+                    (workerThread+                           request+                           coherent_worker)+                    for_worker_thread+                )+                (+                    (   \ _ ->  do+                        -- Actions to take when the thread breaks....+                        writeChan session_input InternalAbort_SIC+                    )+                    :: HTTP500PrecursorException -> IO ()+                )++            H.insert stream2workerthread stream_id thread_id++        return ()+    else+        -- Frame doesn't end the headers... it was added before... so+        -- probably do nothing+        return ()+++clientProcessIncomingHeaders :: NH2.Frame ->  ReaderT SessionData IO ()+clientProcessIncomingHeaders frame | Just (stream_id, bytes) <- isAboutHeaders frame = do++    opens_stream              <- appendHeaderFragmentBlock stream_id bytes+    receiving_headers_mvar    <- view receivingHeaders+    last_good_stream_mvar     <- view lastGoodStream+    decode_headers_table_mvar <- view toDecodeHeaders+    stream_request_headers    <- view stream2HeaderBlockFragment+    response2waiter           <- view (simpleClient . response2Waiter_ClS)++    if opens_stream+      then do+        maybe_rcv_headers_of <- liftIO $ takeMVar receiving_headers_mvar+        case maybe_rcv_headers_of of+          Just _ -> do+            -- Bad peer, 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+            stream_initiated_by_client <- streamInitiatedByClient stream_id+            if (odd stream_id ) && stream_initiated_by_client+              then do+                -- We are golden, set the new good stream+                liftIO $ putMVar last_good_stream_mvar (stream_id)+              else do+                -- We are not golden+                -- TODO: Control for pushed streams here.+                closeConnectionBecauseIsInvalid NH2.ProtocolError+        -- So, now we know, we are ignoring priority frames at the moment+        -- liftIO . logit $ "Priority: " `mappend` (pack . show $ getHeadersPriority frame)+      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 )+        -- Lets get a time+        -- headers_arrived_time      <- liftIO $ getTime Monotonic+        -- Let's decode the headers+        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 <- validateIncomingHeadersClient headers_editor++        good_headers <- case maybe_good_headers_editor of+            Just yes_they_are_good -> return yes_they_are_good+            Nothing -> closeConnectionBecauseIsInvalid NH2.ProtocolError+++        -- If the headers end the request....+        post_data_source <- if not (frameEndsStream frame)+          then do+            mechanism <-  createMechanismForStream stream_id+            return $ postDataSourceFromMechanism mechanism+          else+            return $ return ()++        (Just response_mvar) <- liftIO $ H.lookup response2waiter stream_id+        liftIO $ putMVar response_mvar $ Right (He.toList good_headers, post_data_source)++        return ()+    else+        -- Frame doesn't end the headers... it was added before... so+        -- probably do nothing+        return ()+++streamInitiatedByClient :: GlobalStreamId -> ReaderT SessionData IO Bool+streamInitiatedByClient stream_id = do+    response2waiter <- view (simpleClient . response2Waiter_ClS)+    response_handle_maybe <- liftIO $ H.lookup response2waiter stream_id+    return $ isJust response_handle_maybe++ sendOutFrame :: NH2.EncodeInfo -> NH2.FramePayload  -> ReaderT SessionData IO () sendOutFrame encode_info payload = do     session_output_mvar <- view sessionOutput@@ -730,8 +1040,9 @@         else return headers_editor  -validateIncomingHeaders :: He.HeaderEditor -> ReaderT SessionData IO (Maybe He.HeaderEditor)-validateIncomingHeaders headers_editor = do+-- TODO: Close connection on unexepcted pseudo-headers+validateIncomingHeadersServer :: He.HeaderEditor -> ReaderT SessionData IO (Maybe He.HeaderEditor)+validateIncomingHeadersServer 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...     --@@ -757,7 +1068,27 @@         else             return Nothing +validateIncomingHeadersClient :: He.HeaderEditor -> ReaderT SessionData IO (Maybe He.HeaderEditor)+validateIncomingHeadersClient 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_status = h1 ^. (He.headerLens ":status") +    if (isJust maybe_status)+        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@@ -791,7 +1122,7 @@         -- that it stops accepting frames from connected sources         -- (Streams?)         session_output <- takeMVar session_output_mvar-        writeChan session_output $ Left CancelSession_SOC+        writeChan session_output $ Left (CancelSession_SOC error_code)         putMVar session_output_mvar session_output          -- And unwind the input thread in the session, so that the@@ -799,6 +1130,59 @@         E.throw HTTP2ProtocolException  +-- Sends a GO_AWAY frame and raises an exception, effectively terminating the input+-- thread of the session. This one for the client is different because it throws+-- an exception of type ClientSessionAbortedException+closeConnectionForClient :: NH2.ErrorCodeId -> ReaderT SessionData IO a+closeConnectionForClient error_code = do+    let+        use_reason = case error_code of+            NH2.NoError -> NormalTermination_CCR+            _           -> ProtocolError_CCR++    -- 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+            0+            Nothing+        )+        (NH2.GoAwayFrame+            last_good_stream+            error_code+            ""+        )++    client_is_closed_mvar <- view (simpleClient . clientIsClosed_ClS )++    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 $ error_code+        putMVar session_output_mvar session_output++        -- Let's also mark the session as closed from the client side, so that+        -- any further requests end with the correct exception+        modifyMVar_ client_is_closed_mvar (\ _ -> return True)++        -- And unwind the input thread in the session, so that the+        -- exception handler runs....+        E.throw $ ClientSessionAbortedException use_reason++ frameEndsStream :: InputFrame -> Bool frameEndsStream (NH2.Frame (NH2.FrameHeader _ flags _) _)  = NH2.testEndStream flags @@ -839,7 +1223,10 @@     case pim_maybe of          Just (PostInputMechanism (chan, _))  ->-            liftIO $ putMVar chan Nothing+            liftIO $ do+                putMVar chan Nothing+                -- Not sure if this will work+                H.delete  stream2postinputmechanism stream_id          Nothing ->             -- TODO: This is a protocol error, handle it properly@@ -929,6 +1316,37 @@      -- Pieces of the header     let+       effects              = principal_stream ^. effect_PS+       interrupt_maybe      = effects          ^. interrupt_Ef++++    case interrupt_maybe of++        Nothing -> do+            normallyHandleStream principal_stream++        Just (InterruptConnectionAfter_IEf) -> do+            normallyHandleStream principal_stream+            liftIO . writeChan headers_output $ GoAway_HM (stream_id, effects)++        Just (InterruptConnectionNow_IEf) -> do+            -- Not one hundred-percent sure of this being correct, but we don't want+            -- to acknowledge reception of this stream then+            let use_stream_id = stream_id - 1+            liftIO . writeChan headers_output $ GoAway_HM (use_stream_id, effects)+++normallyHandleStream :: PrincipalStream -> WorkerMonad ()+normallyHandleStream principal_stream = do+    headers_output <- view headersOutput+    stream_id      <- view streamId+    session_settings_mvar <- view sessionSettings_WTE+    session_settings <- liftIO $ readMVar session_settings_mvar+    next_push_stream_mvar <-  view nextPushStream_WTE++    -- Pieces of the header+    let        headers              = principal_stream ^. headers_PS        data_and_conclusion  = principal_stream ^. dataAndConclusion_PS        effects              = principal_stream ^. effect_PS@@ -936,6 +1354,8 @@         can_push             = session_settings ^. pushEnabled +      -- This gets executed in normal conditions, when no interruption is required.+     -- There are several possible moments where the PUSH_PROMISEs can be sent,     -- but a default safe one is before sending the response HEADERS, so that     -- LINK headers in the response come after any potential promises.@@ -1060,10 +1480,13 @@             Nothing ->                 -- This is how we finish sending data                 liftIO $ putMVar data_output (stream_id, Nothing, effect)-            Just bytes -> do-                liftIO $ do-                    putMVar data_output (stream_id, Just bytes, effect)-                consumer data_output+            Just bytes+                | B.length bytes > 0 -> do+                    liftIO $ do+                        putMVar data_output (stream_id, Just bytes, effect)+                    consumer data_output+                | otherwise ->+                    liftIO $ putMVar data_output (stream_id, Nothing, effect)   -- Returns if the frame is the first in the stream@@ -1121,8 +1544,8 @@      use_chunk_length <- view $ sessionsContext .  sessionsConfig . dataFrameSize -    header_output_request <- liftIO $ readChan input_chan-    case header_output_request of+    header_output_request <- {-# SCC input_chan  #-} liftIO $ readChan input_chan+    {-# SCC case_ #-} case header_output_request of         NormalResponse_HM (stream_id, headers_ready_mvar, headers, effect)  -> do              -- First encode the headers using the table@@ -1148,6 +1571,15 @@                     putMVar headers_ready_mvar HeadersSent                     ) +        GoAway_HM (stream_id, _effect) -> do+           -- This is in charge of sending an interrupt message to the framer+           let+               message = Left (SpecificTerminate_SOC stream_id)+           liftIO . withMVar session_output_mvar $+               (\session_output -> do+                   writeChan session_output message+               )+         PushPromise_HM (parent_stream_id, child_stream_id, promise_headers, effect) -> do              encode_dyn_table_mvar <- view toEncodeHeaders@@ -1287,14 +1719,111 @@     writeContinuations fragments stream_id effect =       mapM_ (\ fragment ->                   withLockedSessionOutput-                      (\ session_output -> writeChan session_output $ Right (-                           NH2.EncodeInfo {-                               NH2.encodeFlags     = NH2.defaultFlags-                               ,NH2.encodeStreamId = stream_id-                               ,NH2.encodePadding  = Nothing-                               },-                           NH2.DataFrame fragment,-                           effect )+                      (\ session_output -> do+                             -- TODO:+                             when (B.length fragment == use_chunk_length) $ do+                                 -- Force tons and tons of ping frames to see if we get less delays due to+                                 -- congestion...+                                 writeChan session_output $ Right (+                                     NH2.EncodeInfo {+                                         NH2.encodeFlags     = NH2.defaultFlags+                                         ,NH2.encodeStreamId = 0+                                         ,NH2.encodePadding  = Nothing+                                         },+                                     NH2.PingFrame "talk  on",+                                     effect )++                             writeChan session_output $ Right (+                                 NH2.EncodeInfo {+                                     NH2.encodeFlags     = NH2.defaultFlags+                                     ,NH2.encodeStreamId = stream_id+                                     ,NH2.encodePadding  = Nothing+                                     },+                                 NH2.DataFrame fragment,+                                 effect )                       )              )              fragments+++sessionPollThread :: SessionData -> Chan HeaderOutputMessage -> IO ()+sessionPollThread  session_data headers_output = do+    let+        pending_requests_mvar  = session_data ^. (simpleClient . pendingRequests_ClS)+        client_next_stream_mvar  = session_data ^. (simpleClient . nextStream_ClS)+        response2waiter       = session_data ^. (simpleClient . response2Waiter_ClS)+        for_worker_thread  = session_data ^. forWorkerThread+        session_input  = session_data ^. sessionInput++    ((headers, input_data_stream), response_mvar) <- takeMVar pending_requests_mvar++    new_stream_id <- modifyMVar client_next_stream_mvar (\ n -> return (n + 1, n) )+    let+        worker_environment = set streamId new_stream_id for_worker_thread+        effects = defaultEffects++    -- Store the response place+    new_stream_id `seq` H.insert response2waiter new_stream_id response_mvar++    -- Start by sending the headers of the request+    headers_sent <- liftIO  newEmptyMVar+    liftIO $ writeChan headers_output $ NormalResponse_HM (new_stream_id, headers_sent, headers, effects)++    liftIO . forkIO $ E.catch+        (runReaderT+            (clientWorkerThread new_stream_id effects headers_sent input_data_stream )+            worker_environment+        )+        (+            (   \ _ ->  do+                -- Actions to take when the thread breaks....+                writeChan session_input InternalAbort_SIC+            )+            :: HTTP500PrecursorException -> IO ()+        )++    sessionPollThread session_data headers_output+++clientWorkerThread :: GlobalStreamId -> Effect -> MVar HeadersSent -> InputDataStream -> WorkerMonad ()+clientWorkerThread stream_id effects headers_sent input_data_stream = do+    -- And now work on sending the data, if any...+    runConduit $+        transPipe liftIO input_data_stream+        `fuseBothMaybe`+        sendDataOfStream stream_id headers_sent effects+    return ()+++clientSideTerminate  :: ClientState -> ConnectionCloseReason  -> IO ()+clientSideTerminate client_state reason = do+    let+        terminateOnQueue :: ConnectionCloseReason -> MVar (Message, MVar RequestResult) -> IO ()+        terminateOnQueue reason' q = do+            w <- tryTakeMVar q+            case w of+                Just (_, m) -> do+                    tryPutMVar m (Left reason')+                    return ()++                Nothing ->+                    return ()+++        terminate :: ConnectionCloseReason -> IO ()+        terminate reason' = do+            terminateOnQueue reason' (client_state ^. pendingRequests_ClS)++            -- Kill any waiters in the map+            H.mapM_+                (\(_k,v) -> tryPutMVar v (Left reason))+                (client_state ^. response2Waiter_ClS)++        client_is_closed_mvar = client_state ^. clientIsClosed_ClS++    -- Let's also mark the session as closed from the client side, so that+    -- any further requests end with the correct exception+    modifyMVar_ client_is_closed_mvar $ \ client_is_closed -> do+            unless client_is_closed $+                terminate reason+            return True

@@ -0,0 +1,35 @@+{-# LANGUAGE OverloadedStrings, TemplateHaskell #-}+-- TODO: Remove this file, it is not needed!!+module SecondTransfer.Http2.SimpleClient (+    ClientState(..)  -- reduced functionality is exported to the user, but+                     -- the framework may need a lot.+    ,pendingRequests_ClS+                                   ) where+++import           Control.Lens+import           Control.Concurrent.Chan+-- 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,unless, when, mapM_, forM, forM_)+import           Control.Monad.IO.Class                 (liftIO)+import           Control.DeepSeq                        ( ($!!), deepseq )+import           Control.Monad.Trans.Reader+-- import           Control.Monad.Catch                    (throwM)+import           Control.Concurrent.MVar+++import           Data.Conduit+--import++import           SecondTransfer.MainLoop.ClientPetitioner+import           SecondTransfer.MainLoop.CoherentWorker                       (Headers,InputDataStream)+import           SecondTransfer.Http2.Session                                 (ClientState(..),+                                                                               pendingRequests_ClS+                                                                              )++void::()+void = ()

@@ -0,0 +1,15 @@+module SecondTransfer.MainLoop.ClientPetitioner(+    ClientPetitioner(..)+                                               ) where+++--import           Control.Lens+--import           Data.Conduit+import qualified Data.ByteString                                    as B+++import SecondTransfer.MainLoop.CoherentWorker                       (Headers,InputDataStream)+++class ClientPetitioner a where+    request :: a -> Headers -> InputDataStream -> IO (Headers,InputDataStream)

@@ -29,6 +29,7 @@     , TupledPrincipalStream     , TupledRequest     , FragmentDeliveryCallback+    , InterruptEffect(..)      , headers_RQ     , inputData_RQ@@ -47,6 +48,7 @@     , protocol_Pr     , fragmentDeliveryCallback_Ef     , priorityEffect_Ef+    , interrupt_Ef      , defaultEffects     , coherentToAwareWorker@@ -83,7 +85,7 @@ type Headers = [Header]  -- |This is a Source conduit (see Haskell Data.Conduit library from Michael Snoyman)--- that you can use to retrieve the data sent by the client piece-wise.+-- that you can use to retrieve the data sent by the peer piece-wise. type InputDataStream = Source IO B.ByteString  @@ -157,16 +159,30 @@ type FragmentDeliveryCallback = Int -> TimeSpec -> IO ()  +-- | Types of interrupt effects that can be signaled by aware workers. These include whole+--   connection shutdowns and stream resets. In all the cases, the reason given will be+--   NO_ERROR.+data InterruptEffect = InterruptConnectionAfter_IEf   -- ^ Close and send GoAway /after/ this stream finishes delivery+                       |InterruptConnectionNow_IEf    -- ^ Close and send GoAway /without/ delivering this stream.  This implies that+                                                      --   other fields of the PrincipalStream record will be ignored.+                    -- |InterruptThisStream_IEf       -- ^ Just reset this stream, disabled for now.++ -- | Sometimes a response needs to be handled a bit specially, --   for example by reporting delivery details back to the worker data Effect = Effect {+  -- | A callback to be called whenever a data-packet for this stream is called.   _fragmentDeliveryCallback_Ef :: Maybe FragmentDeliveryCallback -  -- In certain circunstances a stream can use an internal priority,+  -- | In certain circunstances a stream can use an internal priority,   -- not given by the browser and the protocol. Lowest values here are   -- given more priority. Default (when Nothing) is given zero. Cases   -- with negative numbers also work.   ,_priorityEffect_Ef :: Maybe Int++  -- | There are situations when it is desirable to close a stream or the entire+  --   connection. Use this member to indicate that.+  ,_interrupt_Ef :: Maybe InterruptEffect   }  makeLenses ''Effect@@ -174,13 +190,14 @@ defaultEffects :: Effect defaultEffects = Effect {   _fragmentDeliveryCallback_Ef = Nothing,-  _priorityEffect_Ef = Nothing+  _priorityEffect_Ef = Nothing,+  _interrupt_Ef = Nothing    }   -- | You use this type to answer a request. The `Headers` are thus response --   headers and they should contain the :status pseudo-header. The `PushedStreams`---   is a list of pushed streams...(I don't thaink that I'm handling those yet)+--   is a list of pushed streams... they will be pushed to the client. data PrincipalStream = PrincipalStream {   _headers_PS              :: Headers,   _pushedStreams_PS        :: PushedStreams,

@@ -301,7 +301,7 @@ smallWaitTime = 50000  -- provideActions :: Wired_Ptr -> IO (LB.ByteString -> IO (), Int -> IO B.ByteString, IO ())-provideActions :: Wired_Ptr -> IO AttendantCallbacks+provideActions :: Wired_Ptr -> IO IOCallbacks provideActions wired_ptr = do     can_write_mvar <- newMVar True     can_read_mvar  <- newMVar True@@ -370,9 +370,9 @@                       else                           return (False,False) -    return  AttendantCallbacks {-        _pushAction_AtC = pushAction,-        _pullAction_AtC = pullAction,-        _closeAction_AtC = closeAction,-        _bestEffortPullAction_AtC = bestEffortPullAction+    return  IOCallbacks {+        _pushAction_IOC = pushAction,+        _pullAction_IOC = pullAction,+        _closeAction_IOC = closeAction,+        _bestEffortPullAction_IOC = bestEffortPullAction     }

@@ -3,14 +3,15 @@ module SecondTransfer.MainLoop.PushPullType (     PushAction     ,PullAction+    ,BestEffortPullAction     ,Attendant     ,CloseAction-    ,AttendantCallbacks(..)+    ,IOCallbacks(..) -    ,pushAction_AtC-    ,pullAction_AtC-    ,closeAction_AtC-    ,bestEffortPullAction_AtC+    ,pushAction_IOC+    ,pullAction_IOC+    ,closeAction_IOC+    ,bestEffortPullAction_IOC     ) where  @@ -49,22 +50,22 @@  -- | A set of functions describing how to do I/O in a session. --   As usual, we provide lenses accessors.-data AttendantCallbacks = AttendantCallbacks {+data IOCallbacks = IOCallbacks {     -- | put some data in the channel-    _pushAction_AtC               :: PushAction,+    _pushAction_IOC               :: PushAction,     -- | get exactly this much data from the channel. This function can     --   be used by HTTP/2 since lengths are pretty well built inside the     --   protocoll itself.-    _pullAction_AtC               :: PullAction,+    _pullAction_IOC               :: PullAction,     -- | pull data from the channel, as much as the TCP stack wants to provide.     --   we have no option but use this one when talking HTTP/1.1, where the best     --   way to know the length is to scan until a Content-Length is found.-    _bestEffortPullAction_AtC     :: BestEffortPullAction,+    _bestEffortPullAction_IOC     :: BestEffortPullAction,     -- | this is called when we wish to close the channel.-    _closeAction_AtC              :: CloseAction+    _closeAction_IOC              :: CloseAction     } -makeLenses ''AttendantCallbacks+makeLenses ''IOCallbacks  -- | This is an intermediate type. It represents what you obtain --   by combining something that speaks the protocol and an AwareWorker.@@ -83,4 +84,4 @@ --   This library supplies two of such Attendant factories, --   'SecondTransfer.Http1.http11Attendant' for --   HTTP 1.1 sessions, and 'SecondTransfer.Http2.http2Attendant' for HTTP/2 sessions.-type Attendant = AttendantCallbacks -> IO ()+type Attendant = IOCallbacks -> IO ()

@@ -59,6 +59,7 @@     SessionInputThread_HTTP2SessionComponent     |SessionHeadersOutputThread_HTTP2SessionComponent     |SessionDataOutputThread_HTTP2SessionComponent+    |SessionClientPollThread_HTTP2SessionComponent     |Framer_HTTP2SessionComponent     |Session_HTTP11     deriving Show

@@ -10,7 +10,7 @@ import           Control.Lens            ((^.), makeLenses)  -import            System.Log.Logger+import           System.Log.Logger   @@ -45,7 +45,6 @@         }  - sessionExceptionHandler ::     E.Exception e => SessionComponent -> Int -> SessionsContext -> e -> IO () sessionExceptionHandler session_component session_id sessions_context e =@@ -61,7 +60,9 @@             )     in case maybe_error_callback of         Nothing ->-            errorM component_tag (show e)+            -- errorM component_tag (show e)+            -- When no callback, ignore the exception+            return ()          Just callback ->             callback error_tuple

@@ -7,7 +7,7 @@ -- PVP       summary:      +-+------- breaking API changes --                         | | +----- non-breaking API additions --                         | | | +--- code changes with no API change-version     :              0.6.1.0+version     :              0.7.1.0  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.6.1.0+  tag:      0.7.1.0  library @@ -79,6 +79,8 @@                 , SecondTransfer.MainLoop.Tokens                 , SecondTransfer.MainLoop.Framer                 , SecondTransfer.MainLoop.Logging+                , SecondTransfer.MainLoop.ClientPetitioner+                 , SecondTransfer.Sessions.Internal                  , SecondTransfer.MainLoop.OpenSSL_TLS@@ -87,6 +89,7 @@                 , SecondTransfer.Http2.Framer                 , SecondTransfer.Http2.MakeAttendant                 , SecondTransfer.Http2.Session+                , SecondTransfer.Http2.SimpleClient                  , SecondTransfer.Http1.Session @@ -139,26 +142,25 @@   -- Base language which the package is written in.   default-language: Haskell2010 -  -- Very specific directory with the version of openssl-  -- that I'm using. As of February 2015-- this one is not-  -- commonly installed+  -- NOTE: Very specific directory with the version of openssl+  -- that I'm using. As of February 2015-- openssl 1.0.2 is not+  -- commonly installed. Update this path in your build  +  -- or otherwise your build will be broken.   include-dirs: /opt/openssl-1.0.2/include    c-sources: cbits/tlsinc.c -  -- cc-options: -fPIC -pthread -g -O0   if flag(debug)       cc-options: -O0 -g3       ld-options: -g3 -  -- cc-options: -g3 -O0--  -- ghc-options: -O2 -cpp  -pgmPcpphs  -optP--cpp   ghc-options: -pgmPcpphs  -optP--cpp    extra-libraries: ssl crypto -  -- NOTICE: Please fill-in with an-up-to date library path here+  -- NOTE: Please fill-in with an-up-to date library path here.+  -- It should point to openssl 1.0.2 or greater. See note for+  -- include-dirs above.   extra-lib-dirs: /opt/openssl-1.0.2/lib    include-dirs: macros/@@ -177,6 +179,7 @@   + Test-Suite hunit-tests   type               : exitcode-stdio-1.0   main-is            : hunit_tests.hs@@ -230,9 +233,14 @@                        , SecondTransfer.MainLoop.Tokens                        , SecondTransfer.MainLoop.Framer                        , SecondTransfer.MainLoop.Logging+                       , SecondTransfer.MainLoop.Protocol                        , SecondTransfer.Sessions.Internal+                       , SecondTransfer.MainLoop.ClientPetitioner                        , SecondTransfer.Utils                        , SecondTransfer.Http2.Framer                        , SecondTransfer.Http2.MakeAttendant                        , SecondTransfer.Http2.Session                        , SecondTransfer.Http1.Session+                       , Tests.HTTP1Parse+                       , Tests.HTTP2Session+                       , Tests.Utils

@@ -5,79 +5,203 @@   import           Control.Concurrent-import           Control.Concurrent.MVar-import           Control.Concurrent.Chan+import           Control.Concurrent.STM.TMVar+import           Control.Concurrent.STM.TChan import           Control.Exception-import           Control.Lens            (view, (^.))-import qualified Control.Lens            as L+import           Control.Lens                     (view, (^.), over)+import qualified Control.Lens                     as L+import           Control.Monad.STM                (atomically) -import qualified Network.HTTP2           as NH2-import qualified Network.HPACK           as HP+import qualified Network.HTTP2                    as NH2+import qualified Network.HPACK                    as HP -import qualified Data.ByteString         as B-import qualified Data.ByteString.Lazy    as LB+import qualified Data.ByteString                  as B+import qualified Data.ByteString.Lazy             as LB+import qualified Data.ByteString.Builder          as Bu+import qualified Data.Sequence                    as Sq+import           Data.Sequence                    ( (<|), (|>), ViewL(..), Seq )  import           SecondTransfer.Http2 import           SecondTransfer.Types+import           SecondTransfer.Http2.Session+import           SecondTransfer.Sessions.Config+import           SecondTransfer.Sessions.Internal+import           SecondTransfer.Http2.Framer import           SecondTransfer.MainLoop.Internal+import           Debug.Trace +-- "Internal imports"+import           SecondTransfer.MainLoop.CoherentWorker (defaultEffects) ++data DataContinuityEngine = DataContinuityEngine {+    _incoming_DCE :: Seq B.ByteString+    }+++L.makeLenses ''DataContinuityEngine+++emptyDCE :: DataContinuityEngine+emptyDCE =  DataContinuityEngine  Sq.empty+++dceAddData :: DataContinuityEngine -> B.ByteString -> DataContinuityEngine+dceAddData dce moredata | B.length moredata > 0 = over incoming_DCE ( |>  moredata ) dce+dceAddData dce moredata | otherwise             = dce+++tryGetThisManyData :: DataContinuityEngine -> Int -> Maybe (DataContinuityEngine, B.ByteString)+tryGetThisManyData dce n = let+    go_result = go n "" (dce ^. incoming_DCE)+    go :: Int -> Bu.Builder -> Seq B.ByteString -> Maybe (Seq B.ByteString, Bu.Builder)+    go n bld seq  | Sq.EmptyL <- Sq.viewl seq, n > 0   = Nothing+                  | Sq.EmptyL <- Sq.viewl seq, n == 0  = Just (seq, bld)+                  | lead :< rest <- Sq.viewl seq, n > B.length lead    = go (n - B.length lead) (bld `mappend` Bu.byteString lead) rest+                  | lead :< rest <- Sq.viewl seq, n == B.length lead   = Just ( rest, bld `mappend` Bu.byteString lead)+                  | lead :< rest <- Sq.viewl seq, n < B.length lead    = let+                      (tk, lv) = B.splitAt n lead+                      in Just ( lv <| rest, bld `mappend`  Bu.byteString tk  )+  in case go_result of+    Just (new_dce, bld ) -> Just+        (DataContinuityEngine new_dce,  LB.toStrict . Bu.toLazyByteString $ bld )++    Nothing -> Nothing+++dceIsEmpty :: DataContinuityEngine -> Bool+dceIsEmpty dce = Sq.null $ dce ^. incoming_DCE+++dceGetAllData :: DataContinuityEngine -> (DataContinuityEngine, B.ByteString)+dceGetAllData dce = (emptyDCE, all_together)+  where+    all_together = LB.toStrict . Bu.toLazyByteString .  L.foldMapOf (incoming_DCE . L.folded) Bu.byteString $ dce++ data DecoySession = DecoySession {-    _inputDataChannel   :: Chan B.ByteString-    ,_outputDataChannel :: Chan LB.ByteString-    ,_sessionThread     :: ThreadId-    ,_remainingOutputBit:: MVar B.ByteString-    ,_nh2Settings       :: NH2.Settings-    ,_sessionThrowed    :: MVar Bool-    ,_waiting           :: MVar ThreadId-    ,_encodeHeadersHere :: MVar HP.DynamicTable-    ,_decodeHeadersHere :: MVar HP.DynamicTable+    _inputDataChannel     :: TChan B.ByteString+    ,_outputDataChannel   :: TChan B.ByteString+    ,_sessionThread       :: ThreadId+    ,_remainingOutputBit  :: TMVar B.ByteString+    ,_nh2Settings         :: NH2.Settings+    ,_sessionThrowed      :: TMVar Bool+    ,_waiting             :: TMVar ThreadId+    ,_encodeHeadersHere   :: TMVar HP.DynamicTable+    ,_decodeHeadersHere   :: TMVar HP.DynamicTable++    ,_incomingData        :: TMVar DataContinuityEngine+    ,_clientState         :: ClientState+    -- Execute this when you want to use a powerfull client session+    ,_startClientSessionCallback :: IO ClientState     }   L.makeLenses ''DecoySession  ++-- errorsSessionConfig :: MVar Bool -> SessionsConfig+-- errorsSessionConfig mvar = set (sessionsCallbacks . reportErrorCallback_SC)+--      (Just $ setError mvar) defaultSessionsConfig+++channelsToIOCallbacks :: TChan B.ByteString -> TChan B.ByteString -> TMVar DataContinuityEngine -> IOCallbacks+channelsToIOCallbacks input_data_channel output_data_channel incoming_data_tmvar =+  let+    push_action  :: PushAction+    push_action lazy_bs  =  atomically $+        writeTChan output_data_channel $ LB.toStrict lazy_bs++    pull_action  :: PullAction+    pull_action byte_count = atomically $ do+        incoming_data <- takeTMVar incoming_data_tmvar+        let+            go dce = let+                maybe_stuff = tryGetThisManyData dce byte_count+                in case maybe_stuff of+                    Nothing -> do+                        new_data <- readTChan input_data_channel+                        let+                            new_dce = dceAddData dce new_data+                        go new_dce++                    Just (new_bce, bs) -> do+                        putTMVar incoming_data_tmvar new_bce+                        return bs+        go incoming_data++    best_effort_pull_action :: BestEffortPullAction+    best_effort_pull_action can_block = atomically $ do+        incoming_data <- takeTMVar incoming_data_tmvar+        if dceIsEmpty incoming_data+          then+            if can_block+              then do+                -- Wait for the next item+                new_data <- readTChan input_data_channel+                -- Leave the empty thing alone+                putTMVar incoming_data_tmvar incoming_data+                return new_data+              else do+                putTMVar incoming_data_tmvar incoming_data+                return ""+          else do+              let+                  (new_dce, all_together) = dceGetAllData incoming_data+              putTMVar incoming_data_tmvar new_dce+              return all_together++    close_action :: CloseAction+    close_action =  return ()+  in+    IOCallbacks {+       _pushAction_IOC = push_action,+       _pullAction_IOC = pull_action,+       _closeAction_IOC = close_action,+       _bestEffortPullAction_IOC = best_effort_pull_action+       }++ -- Supposed to be an HTTP/2 attendant createDecoySession :: Attendant -> IO DecoySession createDecoySession attendant = do-    input_data_channel  <- newChan-    output_data_channel <- newChan+    input_data_channel  <- newTChanIO+    output_data_channel <- newTChanIO+    incoming_data_tmvar  <- newTMVarIO emptyDCE+    reverse_data_tmvar <- newTMVarIO emptyDCE+    session_throwed_tmvar <- newTMVarIO False+    client_state <- makeClientState+    client_sessions_context <- makeSessionsContext defaultSessionsConfig     let-        push_action  :: PushAction-        push_action = writeChan output_data_channel-        pull_action  :: PullAction-        pull_action = readChan input_data_channel-        close_action :: CloseAction-        best_effort_pull_action _ = error "NotImplemented"-        close_action =  return ()-        attendant_callbacks = AttendantCallbacks {-            _pushAction_AtC = push_action,-            _pullAction_AtC = pull_action,-            _closeAction_AtC = close_action,-            _bestEffortPullAction_AtC = best_effort_pull_action-            }+        attendant_callbacks = channelsToIOCallbacks input_data_channel output_data_channel incoming_data_tmvar+        client_callbacks = channelsToIOCallbacks output_data_channel input_data_channel reverse_data_tmvar+        start_client_session_callback :: IO ClientState+        start_client_session_callback = do+            wrapSession (ClientState_SP client_state) client_sessions_context client_callbacks+            return client_state +     dtable_for_encoding <- HP.newDynamicTableForEncoding 4096-    dtable_for_encoding_mvar <- newMVar dtable_for_encoding+    dtable_for_encoding_mvar <- newTMVarIO dtable_for_encoding     dtable_for_decoding <- HP.newDynamicTableForDecoding 4096-    dtable_for_decoding_mvar <- newMVar dtable_for_decoding+    dtable_for_decoding_mvar <- newTMVarIO dtable_for_decoding -    waiting_mvar <- newEmptyMVar+    waiting_mvar <- newEmptyTMVarIO      thread_id <- forkIO $ catch         (do-            attendant push_action pull_action close_action+            attendant attendant_callbacks         )         ((\ e -> do             putStrLn $ "Exception: " ++ (show e)-            maybe_waiting <- tryTakeMVar waiting_mvar+            maybe_waiting <- atomically $ tryReadTMVar waiting_mvar             case maybe_waiting of                 Just thread_id -> throwTo thread_id e                 Nothing -> return ()         ):: SomeException -> IO ()) -    remaining_output_bit <- newMVar ""+    remaining_output_bit <- newTMVarIO ""     let session = DecoySession {         _inputDataChannel  = input_data_channel,         _outputDataChannel = output_data_channel,@@ -86,7 +210,11 @@         _nh2Settings       = NH2.defaultSettings,         _waiting           = waiting_mvar,         _encodeHeadersHere = dtable_for_encoding_mvar,-        _decodeHeadersHere = dtable_for_decoding_mvar+        _decodeHeadersHere = dtable_for_decoding_mvar,+        _incomingData      = incoming_data_tmvar,+        _sessionThrowed    = session_throwed_tmvar,+        _clientState       = client_state,+        _startClientSessionCallback = start_client_session_callback         }     return session @@ -97,13 +225,15 @@   -- Send a frame to a session-sendFrameToSession :: DecoySession -> OutputFrame -> IO ()+sendFrameToSession :: DecoySession -> (NH2.EncodeInfo, NH2.FramePayload) -> IO () sendFrameToSession session (encode_info, frame_payload) = do     let         bs_list = NH2.encodeFrameChunks encode_info frame_payload         input_data_channel = session ^. inputDataChannel -    writeChan2Chan input_data_channel bs_list+    atomically $+        mapM_+            (writeTChan input_data_channel) bs_list   -- Read a frame from a session... if possible. It will block@@ -114,22 +244,30 @@     let         output_data_channel = decoy_session ^. outputDataChannel         remaining_output_bit_mvar = decoy_session ^. remainingOutputBit-        pull_action = fmap LB.toStrict $ readChan output_data_channel+        pull_action =  readTChan output_data_channel         settings = decoy_session ^. nh2Settings         waiting_for_read = decoy_session ^. waiting     my_thread_id <- myThreadId-    putMVar waiting_for_read my_thread_id-    remaining_output_bit <- takeMVar remaining_output_bit_mvar-    (packet, rest) <- readNextChunkAndContinue http2FrameLength remaining_output_bit pull_action-    putMVar remaining_output_bit_mvar rest-    takeMVar waiting_for_read+    packet <- atomically $ do+        {-# SCC aT #-} putTMVar waiting_for_read my_thread_id+        remaining_output_bit <- {-# SCC bT #-}  takeTMVar remaining_output_bit_mvar+        (packet, rest) <-  {-# SCC cT #-}  readNextChunkAndContinue http2FrameLength remaining_output_bit pull_action+        {-# SCC dT #-} putTMVar remaining_output_bit_mvar rest+        {-# SCC eT #-} takeTMVar waiting_for_read+        return packet     let         error_or_frame = NH2.decodeFrame settings packet     case error_or_frame of         Left   _      -> return Nothing         Right  frame  -> return $ Just frame +-- Reads all frames from a session and assembles a response. Notice that this is a buggy hack,+-- because continuation frames and such are not considered.+-- T+-- readResponseFromSession :: DecoySession -> NH2.StreamId -> IO (Headers, B.ByteString, [NH2.Frame]) ++ -- Encode headers to send to the session -- TODO: There is an important bug here... we are using the default encoding -- strategy everywhere@@ -138,21 +276,20 @@     do         let             mv = decoy_session ^. encodeHeadersHere-        dtable <- takeMVar mv+        dtable <- atomically $ takeTMVar mv         (dtable', bs ) <- HP.encodeHeader HP.defaultEncodeStrategy dtable headers-        putMVar mv dtable'+        atomically $ putTMVar mv dtable'         return bs  -- Decode headers to receive them from the session decodeHeadersForSession :: DecoySession -> B.ByteString -> IO Headers-decodeHeadersForSession decoy_session bs =-    do-        let-            mv = decoy_session ^. decodeHeadersHere-        dtable <- takeMVar mv-        (dtable', headers) <- HP.decodeHeader dtable bs-        putMVar mv dtable'-        return headers+decodeHeadersForSession decoy_session bs =  do+    let+        mv = decoy_session ^. decodeHeadersHere+    dtable <- atomically $ takeTMVar mv+    (dtable', headers) <- HP.decodeHeader dtable bs+    atomically $ putTMVar mv dtable'+    return headers  -- Send raw data to a session sendRawDataToSession :: DecoySession -> B.ByteString -> IO ()@@ -162,9 +299,10 @@         waiting_for_write = decoy_session ^. waiting      thread_id <- myThreadId-    putMVar waiting_for_write thread_id-    writeChan input_data_channel data_to_send-    takeMVar waiting_for_write+    atomically $ do+        putTMVar waiting_for_write thread_id+        writeTChan input_data_channel data_to_send+        takeTMVar waiting_for_write     return ()  

@@ -0,0 +1,46 @@+{-# LANGUAGE OverloadedStrings #-}+module Tests.HTTP1Parse where+++import qualified Data.ByteString                     as B+import           Data.Maybe                          (isJust)++import           Test.HUnit++import           SecondTransfer                      (Headers)+import           SecondTransfer.Http1.Parse+++testParse :: Test+testParse = TestCase $ do+    let+        headers_text                   = "GET /helo.html HTTP/1.1\r\nHost: www.auther.com \r\n\r\n"+        headers_text2                   = "POST /helo.html HTTP/1.1\r\nHost: www.auther.com \r\nContent-Length: 1\r\n\r\np"+        a0                             = newIncrementalHttp1Parser+        isDone (OnlyHeaders_H1PC _ _)  = True+        isDone _                       = False+        a1                             = addBytes a0 headers_text+        (OnlyHeaders_H1PC h0 leftovers)= a1+        (HeadersAndBody_H1PC h1 cond0 l2) = addBytes a0 headers_text2+        waitForBodyOk (HeadersAndBody_H1PC _ _ _)   = True+        waitForBodyOk _ = False+    assertBool "testParse.IsDone" (isDone a1)+    assertEqual "testParse.NoLeftovers" leftovers ""+    assertEqual "testParse.YesLeftovers" l2 "p"+    assertEqual "testParse.FinishWellSeen" (UseBodyLength_BSC 1) cond0+++testGenerate :: Test+testGenerate = TestCase $ do+    let+        headers_list :: Headers+        headers_list = [+            (":status", "200"),+            ("host", "www.example.com"),+            ("etag", "afrh")+            ]+        fragments = ["hello world"]+        serialized = serializeHTTPResponse headers_list fragments+    assertEqual "testGenerate.1"+        "HTTP/1.1 200 OK\r\ncontent-length: 11\r\netag: afrh\r\nhost: www.example.com\r\n\r\nhello world"+        serialized

@@ -0,0 +1,584 @@+{-# LANGUAGE OverloadedStrings #-}+module Tests.HTTP2Session where+++import           Data.Typeable++import           Control.Concurrent               (threadDelay)+import qualified Control.Concurrent               as C (yield)+import           Control.Concurrent.MVar+import           Control.Exception+import           Control.Lens+import qualified Control.Lens                     as L+import           Control.Monad.IO.Class           (liftIO)+import qualified Network.HTTP2                    as NH2+import           Test.HUnit+import           SecondTransfer.Exception+import           SecondTransfer.Http2             (http2Attendant)+import           SecondTransfer.Sessions+import           SecondTransfer.Test.DecoySession+import           SecondTransfer.Types+import           SecondTransfer.Utils.HTTPHeaders (fetchHeader)+import           SecondTransfer.MainLoop.CoherentWorker (defaultEffects)+import           SecondTransfer.MainLoop.ClientPetitioner++++import           Data.Conduit                     (yield)++++saysHello :: DataAndConclusion+saysHello = do+    yield "Hello world!\ns"+    -- No footers+    return []+++simpleWorker :: AwareWorker+simpleWorker = coherentToAwareWorker . const $ return (+    [+        (":status", "200")+    ],+    [], -- No pushed streams+    saysHello+    )+++erringWorker :: AwareWorker+erringWorker = coherentToAwareWorker . const $ return (+    [+        (":status", "500")+    ],+    [], -- No pushed streams+    saysHello+    )+++abortingWorker :: AwareWorker+abortingWorker req_ = do+    pr1 <- erringWorker req_+    let+        pr2 = L.set (effect_PS . interrupt_Ef) (Just InterruptConnectionAfter_IEf) pr1+    return pr2+++earlyAbortingWorker :: AwareWorker+earlyAbortingWorker req_ = do+    pr1 <- erringWorker req_+    let+        pr2 = L.set (effect_PS . interrupt_Ef) (Just InterruptConnectionNow_IEf) pr1+    return pr2+++data Internal500Exception = Internal500Exception+    deriving (Typeable, Show)++instance Exception Internal500Exception where+    toException = convertHTTP500PrecursorExceptionToException+    fromException = getHTTP500PrecursorExceptionFromException+++throwingWorker :: AwareWorker+throwingWorker _ = throwIO Internal500Exception++throwingWorker2 :: AwareWorker+throwingWorker2  = coherentToAwareWorker . const .  return $ (+    [+        -- These headers will be already sent by the time+        -- the exception is discovered...+        (":status", "200")+    ],+    [], -- No pushed streams+    do+        yield "Error coming down"+        liftIO $ throwIO Internal500Exception+        return []+    )+++simpleRequestHeaders :: Headers+simpleRequestHeaders = [+    (":path", "/"),+    (":authority", "www.example.com"),+    (":scheme", "https"),+    (":method", "GET")+    ]+++badRequestHeaders :: Headers+badRequestHeaders = [+    (":path", "/"),+    (":authority", "www.example.com"),+    (":scheme", "https")+    ]+++setError :: MVar Bool -> ErrorCallback+setError mvar = const $ modifyMVar_ mvar (const $ return True )+++errorForPrefaceOk :: MVar Bool ->  ErrorCallback+errorForPrefaceOk ok_mvar (_, _, some_exception) = do+    let+        maybe_blocked :: Maybe BlockedIndefinitelyOnMVar+        maybe_blocked = fromException some_exception+    case maybe_blocked of+        Just _ -> modifyMVar_ ok_mvar (const . return $ True)+        Nothing -> modifyMVar_ ok_mvar (const . return $ False)+++errorsSessionConfig :: MVar Bool -> SessionsConfig+errorsSessionConfig mvar = set (sessionsCallbacks . reportErrorCallback_SC)+    (Just $ setError mvar) defaultSessionsConfig++errorsSessionConfigForMVar :: MVar Bool -> SessionsConfig+errorsSessionConfigForMVar mvar = set (sessionsCallbacks . reportErrorCallback_SC)+    (Just $ errorForPrefaceOk mvar) defaultSessionsConfig+++-- Test disabled due to irrelevance+-- testPrefaceChecks :: Test+-- testPrefaceChecks = TestCase $ do+--     errors_mvar <- newMVar False+--     sessions_context <- makeSessionsContext (errorsSessionConfigForMVar errors_mvar)+--     let+--         attendant = http2Attendant sessions_context simpleWorker+--     decoy_session <- createDecoySession attendant+--     -- This should work+--     sendRawDataToSession decoy_session "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"+--     threadDelay 1000000+--     error_ok <- readMVar errors_mvar+--     if not error_ok then+--         assertFailure "TypeOfErrorNotOK"+--     else+--         return ()+++testPrefaceChecks2 :: Test+testPrefaceChecks2 = TestCase $ do+    errors_mvar <- newMVar False+    sessions_context <- makeSessionsContext (errorsSessionConfig errors_mvar)+    let+        attendant = http2Attendant sessions_context simpleWorker+    decoy_session <- createDecoySession attendant+    -- This should work+    sendRawDataToSession decoy_session "PRI * HXXP/2.0\r\n\r\nSM\r\n\r\n"+    threadDelay 1000000+    got_error <- readMVar errors_mvar+    if not got_error then do+        assertFailure "Exception didn't raise"+    else+        return ()+++testFirstFrameMustBeSettings :: Test+testFirstFrameMustBeSettings = TestCase $ do+    errors_mvar <- newMVar False+    sessions_context <- makeSessionsContext (errorsSessionConfig errors_mvar)+    let+        attendant = http2Attendant sessions_context simpleWorker+    decoy_session <- createDecoySession attendant+    sendRawDataToSession decoy_session "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"+    maybe_frame <- recvFrameFromSession decoy_session+    case maybe_frame of+        Nothing ->+            assertFailure "Waiting a frame, received none"+        Just (NH2.Frame _ (NH2.SettingsFrame _)) -> -- Ok+            return ()++        _ ->+            assertFailure "Waiting a settings frame, received something else"+++testFirstFrameMustBeSettings2 :: Test+testFirstFrameMustBeSettings2 = TestCase $ do+    errors_mvar <- newMVar False+    sessions_context <- makeSessionsContext (errorsSessionConfig errors_mvar)+    let+        attendant = http2Attendant sessions_context simpleWorker+    decoy_session <- createDecoySession attendant+    sendRawDataToSession decoy_session "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"+    maybe_frame <- recvFrameFromSession decoy_session+    case maybe_frame of+        Nothing ->+            assertFailure "Waiting a frame, received none"+        Just (NH2.Frame _ (NH2.SettingsFrame _)) -> -- Ok+            return ()++        _ ->+            assertFailure "Waiting a settings frame, received something else"++    -- Send a settings frame now+    sendFrameToSession+        decoy_session+        ( (NH2.EncodeInfo NH2.defaultFlags 0 Nothing),+          (NH2.SettingsFrame [])+        )++    -- Check session is alive+    got_error <- readMVar errors_mvar+    if got_error then do+        assertFailure "Exception raised unexpectedly"+    else+        return ()++testFirstFrameMustBeSettings3 :: Test+testFirstFrameMustBeSettings3 = TestCase $ do+    errors_mvar <- newMVar False+    sessions_context <- makeSessionsContext (errorsSessionConfig errors_mvar)+    let+        attendant = http2Attendant sessions_context simpleWorker+    decoy_session <- createDecoySession attendant+    sendRawDataToSession decoy_session "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"++    let d2 = ( (NH2.EncodeInfo NH2.defaultFlags 0 Nothing),+          (NH2.PingFrame "01234567") )++    -- Send a ping frame now, so that we get an error+    sendFrameToSession+        decoy_session+        d2++    -- We need to give some time to the framework to react to problems+    threadDelay 20000++    -- Check session is alive+    got_error <- readMVar errors_mvar+    if not got_error then do+        assertFailure "Exception didn't raise properly"+    else+        return ()+++testIGet500Status :: Test+testIGet500Status = TestCase $ do+    errors_mvar <- newMVar False+    sessions_context <- makeSessionsContext (errorsSessionConfig errors_mvar)+    let+        attendant = http2Attendant sessions_context throwingWorker+    decoy_session <- createDecoySession attendant++    -- Send the prologue+    sendRawDataToSession decoy_session "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"++    -- Send a settings frame now, otherwise the session will bark....+    sendFrameToSession+        decoy_session+        ( (NH2.EncodeInfo NH2.defaultFlags 0 Nothing),+          (NH2.SettingsFrame []) )++    -- Now perform a simple, mocking request+    performRequestSimple decoy_session 1 [+        (":method", "get"),+        (":scheme", "https"),+        (":authority", "www.example.com"),+        (":path", "/hi")+        ]++    -- Now we read a few frames+    seen <- return False+    f0 <- recvFrameFromSession decoy_session+    seen1 <- frameIsStatus500 decoy_session seen f0++    f1 <- recvFrameFromSession decoy_session+    seen2 <- frameIsStatus500 decoy_session seen f1++    f2 <- recvFrameFromSession decoy_session+    seen3 <- frameIsStatus500 decoy_session seen f2++    if not seen3 then do+        assertFailure "Didn't see that 500"+    else+        return ()+++frameIsStatus500 :: DecoySession -> Bool -> Maybe NH2.Frame -> IO Bool+frameIsStatus500 decoy_session prev maybe_frame =+    case prev of+        True -> return True+        False ->+            case maybe_frame of+                Just (NH2.Frame _  (NH2.HeadersFrame _ bs ) ) -> do+                    headers <- decodeHeadersForSession decoy_session bs+                    let+                        maybe_status = fetchHeader headers ":status"+                    case maybe_status of+                        Just x | x == "500" -> return True+                        _                   -> return False++                _ ->+                    return False+++testSessionBreaksOnLateError :: Test+testSessionBreaksOnLateError = TestCase $ do+    errors_mvar <- newMVar False+    sessions_context <- makeSessionsContext (errorsSessionConfig errors_mvar)+    let+        attendant = http2Attendant sessions_context throwingWorker2+    decoy_session <- createDecoySession attendant++    -- Send the prologue+    sendRawDataToSession decoy_session "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"++    -- Send a settings frame now, otherwise the session will bark....+    sendFrameToSession+        decoy_session+        ( (NH2.EncodeInfo NH2.defaultFlags 0 Nothing),+          (NH2.SettingsFrame []) )++    -- Now perform a simple, mocking request+    performRequestSimple decoy_session 1 [+        (":method", "get"),+        (":scheme", "https"),+        (":authority", "www.example.com"),+        (":path", "/hi")+        ]++    -- Now we read a few frames+    seen <- return False+    f0 <- recvFrameFromSession decoy_session+    seen1 <- frameIsGoAwayBecauseInternalError decoy_session seen f0++    f1 <- recvFrameFromSession decoy_session+    seen2 <- frameIsGoAwayBecauseInternalError decoy_session seen1 f1++    f2 <- recvFrameFromSession decoy_session+    seen3 <- frameIsGoAwayBecauseInternalError decoy_session seen2 f2++    f3 <- recvFrameFromSession decoy_session+    seen4 <- frameIsGoAwayBecauseInternalError decoy_session seen3 f3++    if not seen4 then do+        assertFailure "Didn't see GoAwayFrame"+    else+        return ()+++frameIsGoAwayBecauseInternalError :: DecoySession -> Bool -> Maybe NH2.Frame -> IO Bool+frameIsGoAwayBecauseInternalError decoy_session prev maybe_frame = do+    case prev of+        True -> return True+        False ->+            case maybe_frame of+                Just (NH2.Frame _  (NH2.GoAwayFrame _ ec _) ) -> do+                    case ec of+                        NH2.InternalError   -> return True+                        _                   -> return False++                _ ->+                    return False+++frameIsGoAwayBecauseProtocolError :: DecoySession -> Bool -> Maybe NH2.Frame -> IO Bool+frameIsGoAwayBecauseProtocolError decoy_session prev maybe_frame = do+    case prev of+        True -> return True+        False ->+            case maybe_frame of+                Just (NH2.Frame _  (NH2.GoAwayFrame _ ec _) ) -> do+                    case ec of+                        NH2.ProtocolError   -> return True+                        _                   -> return False++                _ ->+                    return False+++testUpdateWindowFrameAborts :: Test+testUpdateWindowFrameAborts = TestCase $ do+    errors_mvar <- newMVar False+    sessions_context <- makeSessionsContext (errorsSessionConfig errors_mvar)+    let+        attendant = http2Attendant sessions_context simpleWorker+    decoy_session <- createDecoySession attendant+    sendRawDataToSession decoy_session "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"+    maybe_frame <- recvFrameFromSession decoy_session+    case maybe_frame of+        Nothing ->+            assertFailure "Waiting a frame, received none"+        Just (NH2.Frame _ (NH2.SettingsFrame _)) -> -- Ok+            return ()++        _ ->+            assertFailure "Waiting a settings frame, received something else"++    -- Send a settings frame now+    sendFrameToSession+        decoy_session+        ( NH2.EncodeInfo NH2.defaultFlags 0 Nothing,+          NH2.SettingsFrame [] )++    -- And now send a WindowUpdate frame+    sendFrameToSession+        decoy_session+        (NH2.EncodeInfo NH2.defaultFlags 51 Nothing,+         NH2.WindowUpdateFrame 10 )++    -- Now we read a few frames+    seen <- return False+    f0 <- recvFrameFromSession decoy_session+    seen1 <- frameIsGoAwayBecauseProtocolError decoy_session seen f0++    f1 <- recvFrameFromSession decoy_session+    seen2 <- frameIsGoAwayBecauseInternalError decoy_session seen1 f1++    -- f2 <- recvFrameFromSession decoy_session+    -- seen3 <- frameIsGoAwayBecauseProtocolError decoy_session seen2 f2++    -- f3 <- recvFrameFromSession decoy_session+    -- seen4 <- frameIsGoAwayBecauseProtocolError decoy_session seen3 f3++    if not seen2 then do+        assertFailure "Didn't see GoAwayFrame"+    else+        return ()++    -- Check session is alive+    got_error <- readMVar errors_mvar+    if got_error then do+        assertFailure "Exception raised unexpectedly"+    else+        return ()++testClosedInteraction0 :: Test+testClosedInteraction0 = TestCase $ do+    errors_mvar <- newMVar False+    sessions_context <- makeSessionsContext (errorsSessionConfig errors_mvar)+    let+        attendant = http2Attendant sessions_context simpleWorker+    decoy_session <- createDecoySession attendant+    let+        start_client =  decoy_session ^. startClientSessionCallback+    start_client+    got_error <- readMVar errors_mvar+    if got_error then do+        assertFailure "Exception raised unexpectedly"+    else+        return ()+++testClosedInteraction1 :: Test+testClosedInteraction1 = TestCase $ do+    errors_mvar <- newMVar False+    sessions_context <- makeSessionsContext (errorsSessionConfig errors_mvar)+    let+        attendant = http2Attendant sessions_context simpleWorker+    decoy_session <- createDecoySession attendant+    let+        start_client =  decoy_session ^. startClientSessionCallback+    client_state <- start_client+    got_error <- readMVar errors_mvar+    if got_error then do+        assertFailure "Exception raised unexpectedly"+    else+        return ()++    (headers, input_data_stream) <- request client_state simpleRequestHeaders (return ())+    if length headers <= 0+      then+        assertFailure "NoHeadersBack"+      else+        return ()+    return ()++++testClosedInteraction3 :: Test+testClosedInteraction3 = TestCase $ do+    errors_mvar <- newMVar False+    sessions_context <- makeSessionsContext (errorsSessionConfig errors_mvar)+    let+        attendant = http2Attendant sessions_context simpleWorker+    decoy_session <- createDecoySession attendant+    let+        start_client =  decoy_session ^. startClientSessionCallback+    client_state <- start_client+    ee_mvar <- newMVar False+    catch+        (do+            request client_state badRequestHeaders (return ())+            return ()+        )+        ((\ _ -> modifyMVar_ ee_mvar ( \ _ -> return $ True)  ):: ClientSessionAbortedException -> IO ()  )+    -- if got_error then do+    --     return ()+    -- else+    --     assertFailure "IWasExpectingAnError"+    ee <- takeMVar ee_mvar+    if ee+      then+        return ()+      else+        assertFailure "IWasExpectingAnError--"++++testWorkerClosesAfter :: Test+testWorkerClosesAfter = TestCase $ do+    errors_mvar <- newMVar False+    sessions_context <- makeSessionsContext (errorsSessionConfig errors_mvar)+    let+        attendant = http2Attendant sessions_context abortingWorker+    decoy_session <- createDecoySession attendant+    let+        start_client =  decoy_session ^. startClientSessionCallback+    client_state <- start_client+    got_error <- readMVar errors_mvar+    if got_error then do+        assertFailure "Exception raised unexpectedly"+    else+        return ()++    (headers, _) <- request client_state simpleRequestHeaders (return ())+    if length headers <= 0+      then+        assertFailure "NoHeadersBack"+      else+        return ()++    -- Now, when I try to connect, something bad should happen, since the server must+    -- be closed++    got_error2 <- newMVar False+    catch+        (request client_state simpleRequestHeaders (return ()) >> return () )+        ((\ _ -> modifyMVar_ got_error2 ( \ _ -> return $ True)  )::ClientSessionAbortedException -> IO () )+    ee <- readMVar got_error2+    if ee+      then+        return ()+      else+        assertFailure "Server Must Be Closed"++    return ()++testWorkerClosesBefore :: Test+testWorkerClosesBefore = TestCase $ do+    errors_mvar <- newMVar False+    sessions_context <- makeSessionsContext (errorsSessionConfig errors_mvar)+    let+        attendant = http2Attendant sessions_context earlyAbortingWorker+    decoy_session <- createDecoySession attendant+    let+        start_client =  decoy_session ^. startClientSessionCallback+    client_state <- start_client+    got_error <- readMVar errors_mvar+    if got_error then do+        assertFailure "Exception raised unexpectedly"+    else+        return ()++    got_error2 <- newMVar False+    catch+        (request client_state simpleRequestHeaders (return ()) >> return () )+        ((\ _ -> modifyMVar_ got_error2 ( \ _ -> return $ True)  )::ClientSessionAbortedException -> IO () )+    ee <- readMVar got_error2+    if ee+      then+        return ()+      else+        assertFailure "Server Must Be Closed"++    return ()

@@ -0,0 +1,64 @@+{-# LANGUAGE OverloadedStrings #-}+module Tests.Utils where+++import qualified Data.ByteString                  as B++import           Test.HUnit++import           SecondTransfer                         (Headers)+import           SecondTransfer.Utils.HTTPHeaders +import           SecondTransfer.Http1.Parse    (locateCRLFs)+++testLowercaseHeaders :: Test+testLowercaseHeaders = TestCase $ do +    let +        irregular = [+            ("A", "B"),+            ("C-D", "D")+            ] :: Headers+        regular = lowercaseHeaders irregular+    assertEqual "lowercase" regular [+        ("a", "B"),+        ("c-d", "D")+        ]+    assertEqual "test-lowercase-1" (headersAreLowercase irregular) False+    assertEqual "test-lowercase-2" (headersAreLowercase regular) True+++testCRLFLocate :: Test +testCRLFLocate = TestCase $ do +    let +        input_text = "hello\r\nworld"+        (positions, length_, lastchar) = locateCRLFs 0 [] 0 input_text+    assertEqual "position-is-5.position" (head positions) 5+    assertEqual "position-is-5.length" length_ (B.length input_text)+    assertEqual "position-is-5.lastchar" lastchar (fromIntegral . fromEnum $ 'd')+++    let +        input_text_2 = "hello\r\nworld\r\n\r\n"+        (positions, length_, lastchar) = locateCRLFs 0 [] 0 input_text_2+    assertEqual "many-positions.position" positions [14,12,5]+    assertEqual "many-positions.length" length_ (B.length input_text_2)+    assertEqual "many-positions.lastchar" lastchar (fromIntegral . fromEnum $ '\n')+++testReplaceHostByAuthority :: Test +testReplaceHostByAuthority = TestCase $ do +    let +        headers_pre = [+            ("A", "B"),+            ("C-D", "D"),+            ("Host", "D")+            ] :: Headers+        regular = toList. replaceHostByAuthority . fromList . lowercaseHeaders $ headers_pre+    -- Do not pass in headers with uppercase+    assertEqual "replace-host-by-authority" +        [+            (":authority", "D"),+            ("a", "B"),+            ("c-d", "D")+        ]+        regular

@@ -14,24 +14,29 @@   tests = TestList [-    TestLabel "testPrefaceChecks" testPrefaceChecks,-    TestLabel "testPrefaceChecks2" testPrefaceChecks2,-    TestLabel "testLowercaseHeaders" testLowercaseHeaders,-    TestLabel "testCRLFLocate" testCRLFLocate,-    TestLabel "testHTTP1Parse" testParse,-    TestLabel "testGenerate" testGenerate,-    TestLabel "testReplaceHostByAuthority" testReplaceHostByAuthority,-    TestLabel "testFirstFrameIsSettings" testFirstFrameMustBeSettings,-    TestLabel "testFirstFrameIsSettings2" testFirstFrameMustBeSettings2,-    TestLabel "testFirstFrameIsSettings3" testFirstFrameMustBeSettings3,-    TestLabel "testIGet500Status" testIGet500Status,-    TestLabel "testSessionBreaksOnLateError" testSessionBreaksOnLateError,-    TestLabel "WindowUpdate to unexistent stream" testUpdateWindowFrameAborts+    --TestLabel "testPrefaceChecks" testPrefaceChecks+    TestLabel "testPrefaceChecks2" testPrefaceChecks2+    ,TestLabel "testLowercaseHeaders" testLowercaseHeaders+    ,TestLabel "testCRLFLocate" testCRLFLocate+    ,TestLabel "testHTTP1Parse" testParse+    ,TestLabel "testGenerate" testGenerate+    ,TestLabel "testReplaceHostByAuthority" testReplaceHostByAuthority+    ,TestLabel "testFirstFrameIsSettings" testFirstFrameMustBeSettings+    ,TestLabel "testFirstFrameIsSettings2" testFirstFrameMustBeSettings2+    ,TestLabel "testFirstFrameIsSettings3" testFirstFrameMustBeSettings3+    ,TestLabel "testIGet500Status" testIGet500Status+    ,TestLabel "testSessionBreaksOnLateError" testSessionBreaksOnLateError+    ,TestLabel "WindowUpdate to unexistent stream" testUpdateWindowFrameAborts+    ,TestLabel "testClosedInteraction0" testClosedInteraction0+    ,TestLabel "testClosedInteraction1" testClosedInteraction1+    ,TestLabel "testClosedInteraction3" testClosedInteraction3+    ,TestLabel "testWorkerClosesAfter"  testWorkerClosesAfter+    ,TestLabel "testWorkerClosesBefore" testWorkerClosesBefore     ]  -main = do -    rets <- runTestTT tests +main = do+    rets <- runTestTT tests     if (errors rets == 0 && failures rets == 0)         then exitWith ExitSuccess         else exitWith $ ExitFailure (-1)