first round of changes

servant-server/src/Servant/Server/Internal.hs

@@ -10,6 +10,7 @@
 {-# LANGUAGE PolyKinds                  #-}
 {-# LANGUAGE RankNTypes                 #-}
 {-# LANGUAGE ScopedTypeVariables        #-}
+{-# LANGUAGE TupleSections              #-}
 {-# LANGUAGE TypeFamilies               #-}
 {-# LANGUAGE TypeOperators              #-}
 
@@ -67,8 +68,8 @@ import           Servant.API                 ((:<|>) (..), (:>), BasicAuth, Capt
                                               WithNamedContext,
                                               Description, Summary,
                                               Accept(..),
-                                              Framing(..), Stream,
+                                              FramingRender(..), Stream,
-                                              StreamGenerator(..),
+                                              StreamGenerator(..), ToStreamGenerator(..),
                                               BoundaryStrategy(..))
 import           Servant.API.ContentTypes    (AcceptHeader (..),
                                               AllCTRender (..),
@@ -287,40 +288,68 @@ instance OVERLAPPING_
           status = toEnum . fromInteger $ natVal (Proxy :: Proxy status)
 
 
-instance ( MimeRender ctype a, ReflectMethod method, Framing framing ctype
+instance OVERLAPPABLE_
-         ) => HasServer (Stream method framing ctype a) context where
+         ( MimeRender ctype a, ReflectMethod method,
+           FramingRender framing ctype, ToStreamGenerator f a
+         ) => HasServer (Stream method framing ctype (f a)) context where
 
-  type ServerT (Stream method framing ctype a) m = m (StreamGenerator a)
+  type ServerT (Stream method framing ctype (f a)) m = m (f a)
   hoistServerWithContext _ _ nt s = nt s
 
-  route Proxy _ action = leafRouter $ \env request respond ->
+  route Proxy _ = streamRouter ([],) method (Proxy :: Proxy framing) (Proxy :: Proxy ctype)
+      where method = reflectMethod (Proxy :: Proxy method)
+
+instance OVERLAPPING_
+         ( MimeRender ctype a, ReflectMethod method,
+           FramingRender framing ctype, ToStreamGenerator f a,
+           GetHeaders (Headers h (f a))
+         ) => HasServer (Stream method framing ctype (Headers h (f a))) context where
+
+  type ServerT (Stream method framing ctype (Headers h (f a))) m = m (Headers h (f a))
+  hoistServerWithContext _ _ nt s = nt s
+
+  route Proxy _ = streamRouter (\x -> (getHeaders x, getResponse x)) method (Proxy :: Proxy framing) (Proxy :: Proxy ctype)
+      where method = reflectMethod (Proxy :: Proxy method)
+
+
+streamRouter :: (MimeRender ctype a, FramingRender framing ctype, ToStreamGenerator f a) =>
+                (b -> ([(HeaderName, B.ByteString)], f a))
+             -> Method
+             -> Proxy framing
+             -> Proxy ctype
+             -> Delayed env (Handler b)
+             -> Router env
+streamRouter splitHeaders method framingproxy ctypeproxy action = leafRouter $ \env request respond ->
           let accH    = fromMaybe ct_wildcard $ lookup hAccept $ requestHeaders request
-              cmediatype = NHM.matchAccept [contentType (Proxy :: Proxy ctype)] accH
+              cmediatype = NHM.matchAccept [contentType ctypeproxy] accH
               accCheck = when (isNothing cmediatype) $ delayedFail err406
               contentHeader = (hContentType, NHM.renderHeader . maybeToList $ cmediatype)
           in runAction (action `addMethodCheck` methodCheck method request
                                `addAcceptCheck` accCheck
-                       ) env request respond $ \ (StreamGenerator k) ->
+                       ) env request respond $ \ output ->
-                Route $ responseStream status200 [contentHeader] $ \write flush -> do
+                let (headers, fa) = splitHeaders output
-                      write . BB.lazyByteString . header (Proxy :: Proxy framing) $ (Proxy :: Proxy ctype)
+                    k = getStreamGenerator . toStreamGenerator $ fa in
-                      case boundary (Proxy :: Proxy framing) (Proxy :: Proxy ctype) of
+                Route $ responseStream status200 (contentHeader : headers) $ \write flush -> do
+                      write . BB.lazyByteString $ header framingproxy ctypeproxy
+                      case boundary framingproxy ctypeproxy of
                            BoundaryStrategyBracket f ->
-                                    let go x = let bs = mimeRender (Proxy :: Proxy ctype) $ x
+                                    let go x = let bs = mimeRender ctypeproxy $ x
                                                    (before, after) = f bs
                                                in write (   BB.lazyByteString before
                                                          <> BB.lazyByteString bs
-                                                         <> BB.lazyByteString after)
+                                                         <> BB.lazyByteString after) >> flush
                                     in k go go
                            BoundaryStrategyIntersperse sep -> k
                              (\x -> do
-                                write . BB.lazyByteString . mimeRender (Proxy :: Proxy ctype) $ x
+                                write . BB.lazyByteString . mimeRender ctypeproxy $ x
                                 flush)
                              (\x -> do
-                                write . (BB.lazyByteString sep <>) . BB.lazyByteString . mimeRender (Proxy :: Proxy ctype) $ x
+                                write . (BB.lazyByteString sep <>) . BB.lazyByteString . mimeRender ctypeproxy $ x
                                 flush)
-                      write . BB.lazyByteString . terminate (Proxy :: Proxy framing) $ (Proxy :: Proxy ctype)
+                           BoundaryStrategyGeneral f ->
-    where method = reflectMethod (Proxy :: Proxy method)
+                                    let go = (>> flush) . write . BB.lazyByteString . f . mimeRender ctypeproxy
-
+                                    in k go go
+                      write . BB.lazyByteString $ terminate framingproxy ctypeproxy
 
 -- | If you use 'Header' in one of the endpoints for your API,
 -- this automatically requires your server-side handler to be a function

servant/src/Servant/API.hs

@@ -83,8 +83,9 @@ import           Servant.API.IsSecure        (IsSecure (..))
 import           Servant.API.QueryParam      (QueryFlag, QueryParam,
                                               QueryParams)
 import           Servant.API.Raw             (Raw)
-import           Servant.API.Stream          (Stream, StreamGenerator(..),
+import           Servant.API.Stream          (Stream, StreamGenerator(..), ToStreamGenerator(..),
-                                              Framing(..), BoundaryStrategy(..),
+                                              FramingRender(..), BoundaryStrategy(..),
+                                              FramingUnrender(..),
                                               NewlineFraming)
 import           Servant.API.RemoteHost      (RemoteHost)
 import           Servant.API.ReqBody         (ReqBody)

servant/src/Servant/API/Stream.hs

@@ -5,38 +5,74 @@
 {-# LANGUAGE MultiParamTypeClasses    #-}
 {-# LANGUAGE OverloadedStrings        #-}
 {-# LANGUAGE PolyKinds                #-}
+{-# LANGUAGE TupleSections            #-}
 {-# OPTIONS_HADDOCK not-home          #-}
 
 module Servant.API.Stream where
 
+import           Control.Arrow               ((***), first)
 import           Data.ByteString.Lazy        (ByteString, empty)
+import qualified Data.ByteString.Lazy.Char8  as LB
 import           Data.Proxy                  (Proxy)
 import           Data.Typeable               (Typeable)
 import           GHC.Generics                (Generic)
+import           Text.Read                   (readMaybe)
 
 -- | A stream endpoint for a given method emits a stream of encoded values at a given Content-Type, delimited by a framing strategy.
 data Stream (method :: k1) (framing :: *) (contentType :: *) a
   deriving (Typeable, Generic)
 
 -- | Stream endpoints may be implemented as producing a @StreamGenerator@ -- a function that itself takes two emit functions -- the first to be used on the first value the stream emits, and the second to be used on all subsequent values (to allow interspersed framing strategies such as comma separation).
-data StreamGenerator a =  StreamGenerator ((a -> IO ()) -> (a -> IO ()) -> IO ())
+newtype StreamGenerator a =  StreamGenerator {getStreamGenerator :: (a -> IO ()) -> (a -> IO ()) -> IO ()}
 
--- | The Framing class provides the logic for each framing strategy. The strategy emits a header, followed by boundary-delimited data, and finally a termination character. For many strategies, some of these will just be empty bytestrings.
+-- | ToStreamGenerator is intended to be implemented for types such as Conduit, Pipe, etc. By implementing this class, all such streaming abstractions can be used directly as endpoints.
-class Framing strategy a where
+class ToStreamGenerator f a where
+   toStreamGenerator :: f a -> StreamGenerator a
+
+instance ToStreamGenerator StreamGenerator a
+   where toStreamGenerator x = x
+
+-- | The FramingRender class provides the logic for emitting a framing strategy. The strategy emits a header, followed by boundary-delimited data, and finally a termination character. For many strategies, some of these will just be empty bytestrings.
+class FramingRender strategy a where
    header    :: Proxy strategy -> Proxy a -> ByteString
    boundary  :: Proxy strategy -> Proxy a -> BoundaryStrategy
    terminate :: Proxy strategy -> Proxy a -> ByteString
 
 -- | The bracketing strategy generates things to precede and follow the content, as with netstrings.
--- | The intersperse strategy inserts seperators between things, as with newline framing.
+--   The intersperse strategy inserts seperators between things, as with newline framing.
+--   Finally, the general strategy performs an arbitrary rewrite on the content, to allow escaping rules and such.
 data BoundaryStrategy = BoundaryStrategyBracket (ByteString -> (ByteString,ByteString))
                       | BoundaryStrategyIntersperse ByteString
+                      | BoundaryStrategyGeneral (ByteString -> ByteString)
 
+-- | The FramingUnrender class provides the logic for parsing a framing strategy. Given a ByteString, it strips the header, and returns a tuple of the remainder along with a step function that can progressively "uncons" elements from this remainder. The error state is presented per-frame so that protocols that can resume after errors are able to do so.
+
+class FramingUnrender strategy a where
+   unrenderFrames :: Proxy strategy -> Proxy a -> ByteString -> (ByteString,  ByteString -> (Either String ByteString, ByteString))
 
 -- | A simple framing strategy that has no header or termination, and inserts a newline character between each frame.
+--   This assumes that it is used with a Content-Type that encodes without newlines (e.g. JSON).
 data NewlineFraming
 
-instance Framing NewlineFraming a where
+instance FramingRender NewlineFraming a where
    header    _ _ = empty
    boundary  _ _ = BoundaryStrategyIntersperse "\n"
    terminate _ _ = empty
+
+instance FramingUnrender NewlineFraming a where
+   unrenderFrames _ _ = (, (Right *** LB.drop 1) . LB.break (== '\n'))
+
+-- | The netstring framing strategy as defined by djb: <http://cr.yp.to/proto/netstrings.txt>
+data NetstringFraming
+
+instance FramingRender NetstringFraming a where
+   header    _ _ = empty
+   boundary  _ _ = BoundaryStrategyBracket $ \b -> (LB.pack . show . LB.length $ b, "")
+   terminate _ _ = empty
+
+instance FramingUnrender NetstringFraming a where
+   unrenderFrames _ _ = (, \b -> let (i,r) = LB.break (==':') b
+                                 in case readMaybe (LB.unpack i) of
+                                    Just len -> first Right $ LB.splitAt len . LB.drop 1 $ r
+                                    Nothing -> (Left ("Bad netstring frame, couldn't parse value as integer value: " ++ LB.unpack i), LB.drop 1 . LB.dropWhile (/= ',') $ r)
+                        )

servant/src/Servant/Utils/Links.hs

@@ -117,6 +117,7 @@ import Servant.API.RemoteHost ( RemoteHost )
 import Servant.API.Verbs ( Verb )
 import Servant.API.Sub ( type (:>) )
 import Servant.API.Raw ( Raw )
+import Servant.API.Stream ( Stream )
 import Servant.API.TypeLevel
 import Servant.API.Experimental.Auth ( AuthProtect )
 
@@ -306,6 +307,10 @@ instance HasLink Raw where
     type MkLink Raw = Link
     toLink _ = id
 
+instance HasLink (Stream m fr ct a) where
+    type MkLink (Stream m fr ct a) = Link
+    toLink _ = id
+
 -- AuthProtext instances
 instance HasLink sub => HasLink (AuthProtect tag :> sub) where
   type MkLink (AuthProtect tag :> sub) = MkLink sub