Client support for server-sent events (SSE)

[vapourismo]

This PR is an attempt to implement SSE for the Servant client-side.

---

Small example

import qualified Data.Aeson                           as Aeson
import           Network.HTTP.Client.TLS
import           Servant.API
import           Servant.Client.Core.ServerSentEvents (JsonEventStreamT (..))
import           Servant.Client.Streaming
import           Servant.Types.SourceT                (foreachStep)

data NewsEntry

instance Show NewsEntry

instance Aeson.FromJSON NewsEntry

type NewsAPI = "news-stream" :> ServerSentEvents 'JsonEvent [NewsEntry]

newsAPI :: ClientM (JsonEventStreamT IO [NewsEntry])
newsAPI = client (Proxy @NewsAPI)

main :: IO ()
main = do
  mgr <- newTlsManager
  let env = mkClientEnv mgr $ BaseUrl Https "api.example.com" 443 "stable"
  withClientM newsAPI env $ either print $
    foreachStep fail print . unJsonEventStreamT

[vapourismo]

Ping @haskell-servant/maintainers

[fisx]

Thanks for the PR / sorry for the low bandwidth!

I do not speak for all maintainers, but I think our priorities are #1312, then #1314, then release, then see what's up next, so it may take a while before this gets the attention it deserves.

Is there any reason why this should go to servant-client-core, as opposed to a newly created servant-client-sse? Then you could maintain this yourself until this PR gets handled.

Other maintainers: any thoughts? @vapourismo would you be open to joining us?

[vapourismo]

Thanks for the update. I don’t need this merged right away - so no pressure.
I pinged the maintainers in case it slipped through the cracks.

[hasufell]

I don't see anything blocking this. Why has this not moved forward?

I rebased this branch against master here: https://github.com/hasufell/servant/tree/feature/server-sent-events

[tchoutri]

Yes, ping @haskell-servant/maintainers on this. I personally don't see anything shocking.

[BlastWind]

@vapourismo

Hey Ole, I am running into some trouble trying to use your great work here and would appreciate it if you can point me in the right direction.

My server: I have a server running with an SSE endpoint that spits back strings of DummyObj 3 times, spaced out by 1 second.
My client: I pasted your commit in my code which uses the latest Servant. My code is practically the same as your NewsEntry example, except with the DummyObj object.

My JsonEventStreamT IO DummyObj seems to be only spitting back S.Effect, and never S.Yield. I first slapped traceShow to mimeUnrender, which shows that the client in fact ingest a ByteString representing what I want from the server. I then wanted to find out when this ByteString is being converted into a DummyObj, but when I put a traceShow on DummyObj's parseJSON, it revealed that parseJSON was enver called, is this expected?

Here's my client main function:

main :: IO ()
main = do
  mgr <- newTlsManager
  let env = mkClientEnv mgr $ BaseUrl Http "localhost" 8000 ""
  withClientM cli env $ \case
    Left err -> (print err)
    Right stream -> do 
       putStrLn "operating on stream obj"
       let sourceT = unJsonEventStreamT stream
       S.unSourceT sourceT go
        where 
          go :: StepT IO (Event DummyObj) -> IO ()
          go S.Stop = return ()
          go (S.Error err) = do 
            print "error"
            print err
          go (S.Skip s) = do 
            print "skip"
            go s
          go (S.Effect ms) = do
            print "effect"
            result <- ms
            go result
          go (S.Yield a s) = do 
            putStrLn $ show "yield: " ++ show a 
            hFlush stdout
            go s

I had to modify your code a bit because it wasn't typechecking against the latest version of Servant. Perhaps something incorrect jump out at you in this little snippet already?
I am pattern matching on StepT directly so to get more log messages. Unfortunately, I only get multiple effect logs and then an error log: "Error in $: not enough input"

[vapourismo]

That seems to suggest that the JSON could not be parsed. You'll only get the S.Yield events when the message data body could actually be parsed.

Could it be that the data messages sent by the server are incomplete or are transferred in a chunked way such that you need to stitch multiple messages together?

[voidus]

I'm interested to see this merged, seems to me the only conflict is in context (base upper bound has been raised in servant-client-core.cabal)

[tchoutri]

@voidus Ping me in the PR you open so that I can review & merge. :)

[freckletonj]

It took me a bit to figure out the non-JSON version, so in case it helps anyone:

import Network.HTTP.Client.TLS ( newTlsManager )
import Servant.API
    ( ServerSentEvents, EventKind(RawEvent) )
import Servant.Client.Core.ServerSentEvents (Event (Event), EventStreamT (unEventStreamT))
import Servant.Client.Streaming
    ( ClientM,
      BaseUrl(BaseUrl),
      Scheme(Http),
      client,
      mkClientEnv,
      withClientM )
import Servant.Types.SourceT (StepT (..), SourceT (unSourceT))
import Data.Data (Proxy(..))
import GHC.IO.Handle (hFlush)
import GHC.IO.Handle.FD (stdout)
import qualified Data.ByteString.Char8 as BS

type MyAPI = ServerSentEvents 'RawEvent (Event BS.ByteString)

newsAPI :: ClientM (EventStreamT IO)
newsAPI = client (Proxy @MyAPI)

main :: IO ()
main = do
  mgr <- newTlsManager
  let env = mkClientEnv mgr $ BaseUrl Http "localhost" 8080 ""
  withClientM newsAPI env $ \case
    Left err -> (print err)
    Right stream -> do
       putStrLn "operating on stream obj"
       -- let sourceT = unJsonEventStreamT stream
       let sourceT = unEventStreamT stream
       unSourceT sourceT go
        where
          go :: StepT IO (Event BS.ByteString) -> IO ()
          go Stop = return ()
          go (Error err) = do
            putStrLn "error"
            print err
          go (Skip s) = do
            putStrLn "skip"
            go s
          go (Effect ms) = do
            putStrLn "effect"
            result <- ms
            go result
          go (Yield (Event name dat) s) = do
            putStrLn $ "yield: " ++ BS.unpack dat
            hFlush stdout
            go s

[bflyblue]

Any ideas on how the SSE client and https://hackage.haskell.org/package/servant-event-stream should work together?