Utilize 'StepT' over 'SourceT' for event stream sources

Author: vapourismo

servant-client-core/src/Servant/Client/Core/ServerSentEvents.hs

@@ -48,7 +48,7 @@ import           Servant.API.ContentTypes
 import           Servant.API.Stream
                  (FromSourceIO (..))
 import           Servant.Types.SourceT
-                 (SourceT, StepT (..), foreachYieldStep, mapStepT,
+                 (SourceT (unSourceT), StepT (..), foreachYieldStep, mapStepT,
                  transformStepWithAtto)
 
 -- | Line (or frame) of an event stream
@@ -249,7 +249,7 @@ eventsFromMessages =
 -- See <https://www.w3.org/TR/2009/WD-eventsource-20090421/> for more details.
 --
 newtype EventMessageStreamT m = EventMessageStreamT
-    { unEventMessageStreamT :: SourceT m EventMessage }
+    { unEventMessageStreamT :: StepT m EventMessage }
     deriving (Show, Semigroup, Monoid)
 
 -- | Server-sent event messages
@@ -260,23 +260,27 @@ newtype EventMessageStreamT m = EventMessageStreamT
 instance MonadIO m => FromSourceIO EventStreamChunk (EventMessageStreamT m) where
     fromSourceIO =
         EventMessageStreamT
-        . mapStepT (eventMessagesFromLines . eventLinesFromChunks)
+        . eventMessagesFromLines
+        . eventLinesFromChunks
+        . Effect
+        . ($ pure)
+        . unSourceT
         . fromSourceIO
 
 -- | Server-sent event stream (SSE)
 --
 -- See <https://www.w3.org/TR/2009/WD-eventsource-20090421/> for more details.
 --
 newtype EventStreamT m = EventStreamT
-    { unEventStreamT :: SourceT m (Event ByteString.ByteString) }
+    { unEventStreamT :: StepT m (Event ByteString.ByteString) }
     deriving (Show, Semigroup, Monoid)
 
 -- | Server-sent events
 instance MonadIO m => FromSourceIO EventStreamChunk (EventStreamT m) where
     fromSourceIO input =
         -- 'coerce' is used in place of unpacking and repacking 'EventStreamT'
         coerce
-            (mapStepT eventsFromMessages)
+            eventsFromMessages
             (fromSourceIO input :: EventMessageStreamT m)
 
 -- | Try to parse event data to JSON.
@@ -293,7 +297,7 @@ jsonEventsFromEvents =
 
 -- | Server-sent event stream (SSE) for JSON values
 newtype JsonEventStreamT m a = JsonEventStreamT
-    { unJsonEventStreamT :: SourceT m (Event a) }
+    { unJsonEventStreamT :: StepT m (Event a) }
     deriving (Show, Functor, Semigroup, Monoid)
 
 -- | Server-sent JSON event stream
@@ -302,5 +306,5 @@ instance (MonadIO m, Aeson.FromJSON a) => FromSourceIO EventStreamChunk (JsonEve
         -- The 'coerce' efficiently unwraps the 'EventStreamT' and wraps the
         -- JsonEventStreamT.
         coerce
-            (mapStepT jsonEventsFromEvents)
+            jsonEventsFromEvents
             (fromSourceIO input :: EventStreamT m)

servant-client-core/test/Servant/Client/Core/ServerSentEventsSpec.hs

@@ -17,7 +17,7 @@ import           Servant.Client.Core.ServerSentEvents
                  (Event (..), EventIgnoreReason (EventComment),
                  EventMessage (..), unEventMessageStreamT, unEventStreamT)
 import           Servant.Types.SourceT
-                 (runSourceT, source)
+                 (runStepT, source)
 import           Test.Hspec
                  (Spec, describe, it, shouldBe)
 
@@ -41,7 +41,7 @@ spec = describe "Servant.Client.Core.ServerSentEvent" $ do
             for_ [1, 10, 100] $ \chunkSize -> do
                 result <-
                     runExceptT
-                    $ runSourceT
+                    $ runStepT
                     $ unEventMessageStreamT
                     $ fromSourceIO
                     $ source
@@ -80,7 +80,7 @@ spec = describe "Servant.Client.Core.ServerSentEvent" $ do
             for_ [1, 10, 100] $ \chunkSize -> do
                 result <-
                     runExceptT
-                    $ runSourceT
+                    $ runStepT
                     $ unEventStreamT
                     $ fromSourceIO
                     $ source