Потоковые байты в сетевой веб-сокет

Вот несколько кусочков, которые могут сделать вещи более понятными. Во-первых, для первой небольшой демонстрации, пересматривая ваш getObject, я использую Streaming.ByteString.writeFile, который в любом случае находится в ResourceT, чтобы отбросить обход с помощью ленивой строки байтов.

{-# LANGUAGE OverloadedStrings,ScopedTypeVariables #-}

import qualified Aws
import qualified Aws.S3 as S3
import           Data.Conduit 
import qualified Data.Conduit.List as CL (mapM_)
import qualified Data.ByteString.Streaming.HTTP as HTTP
import qualified Data.ByteString.Streaming as SB
import qualified Data.ByteString.Streaming.Internal as SB
import qualified Data.ByteString as B
import qualified Data.ByteString.Lazy as BL
import           Streaming as S
import           Streaming.Prelude as S hiding (show,print)
import           Control.Concurrent.Async (async,waitCatch)
import           Data.Text as T (Text) 
import qualified Network.WebSockets as WebSockets
import           Control.Monad.Trans.Resource

data AwsConfig a = AwsConfig { _aws_cfg :: Aws.Configuration
                             , _aws_s3cfg :: S3.S3Configuration a
                             , _aws_httpmgr :: HTTP.Manager }

getObject :: AwsConfig Aws.NormalQuery -> FilePath -> T.Text -> T.Text ->  IO Int
getObject cfg file bucket key = do
  req <- waitCatch =<< async (runResourceT $ do
    S3.GetObjectResponse { S3.gorResponse = rsp, S3.gorMetadata = mdata } <- 
      Aws.pureAws (_aws_cfg cfg) (_aws_s3cfg cfg) (_aws_httpmgr cfg) $
        S3.getObject bucket key
    let bytestream = do 
         -- lookup "content-length" (S3.omUserMetadata mdata))
         SB.chunk B.empty -- this will be replaced by content-length 
         hoist lift (HTTP.responseBody rsp)  $$+- CL.mapM_ SB.chunk 
    SB.writeFile file bytestream ) -- this is in ResourceT 
  case req of
    Left _ -> return 2
    Right _ -> return 0

Мы можем более или менее абстрагироваться от того, что вы делали с SB.writeFile:

getObjectAbstracted
      :: (SB.ByteString (ResourceT IO) () -> ResourceT IO b)
         -> AwsConfig Aws.NormalQuery -> S3.Bucket -> Text -> ResourceT IO b
getObjectAbstracted action cfg bucket key = do
    S3.GetObjectResponse { S3.gorResponse = rsp, S3.gorMetadata = mdata } <- 
      Aws.pureAws (_aws_cfg cfg) 
                  (_aws_s3cfg cfg) 
                  (_aws_httpmgr cfg) 
                  (S3.getObject bucket key)

    action (hoist lift (HTTP.responseBody rsp)  $$+- CL.mapM_ SB.chunk) 

Здесь нам нужен небольшой помощник, не включенный в потоковую библиотеку bytestring.

mapMChunks_ :: Monad m => (B.ByteString -> m ()) -> SB.ByteString m r -> m r
mapMChunks_ act bytestream = do
  (a S.:> r) <- SB.foldlChunksM (\_ bs -> act bs) (return ()) bytestream
  return r

и может действовать более или менее так, как планировал @haoformayor, используя потоковую строку байтов

writeConnection :: MonadIO m => WebSockets.Connection -> SB.ByteString m r -> m r
writeConnection connection  = 
  mapMChunks_ (liftIO . WebSockets.sendBinaryData connection)

-- following `haoformayor`
connectWrite
    :: (MonadResource m, WebSockets.WebSocketsData a) 
    => WebSockets.PendingConnection 
    -> a                  -- closing  message
    -> SB.ByteString m r  -- stream from aws
    -> m r
connectWrite request closeMessage bytestream = do
    (releaseKey, connection) <- allocate (WebSockets.acceptRequest request)
                                         (`WebSockets.sendClose` closeMessage)
    writeConnection connection bytestream

getObjectWS :: WebSockets.WebSocketsData a =>
       WebSockets.PendingConnection
       -> a
       -> AwsConfig Aws.NormalQuery
       -> S3.Bucket
       -> Text
       -> ResourceT IO ()
getObjectWS request closeMessage = getObjectAbstracted (connectWrite request closeMessage)

Конечно, ни один из них пока не использует разницу между conduit и streaming/streaming-bytestring.

Вы правы, думая, что ручки потребуют дополнительных хитростей. Однако, поскольку вы уже используете монадный преобразователь ResourceT, это восхитительно просто сделать с allocate. allocate позволяет вам создать дескриптор в монаде ресурса и зарегистрировать действие очистки (которое в вашем случае просто закрывает соединение).

ok <- runResourceT $ do
  (releaseKey, handle) <-
    allocate (WebSockets.acceptRequest request) 
             (`WebSockets.sendClose` closeMessage)
  WebSockets.sendBinaryData handle data
  return ok
where
  request = ...
  closeMessage = ...
  data = ...
  ok = ...

При использовании allocate дескриптор гарантированно закроется к тому времени, когда runResourceT вернет ok.

Однако я не совсем уверен, что это то, чего вы хотите. Мне кажется, что getObject не должен знать о том, как принимать и закрывать WS-соединения; возможно, он должен принять дескриптор соединения WS в качестве аргумента, а затем записать его. Если вы повысите его тип возвращаемого значения до ResourceT, тогда вы можете поручить вызывающему объекту до getObject ответственность за вызов runResourceT и выделение дескрипторов WS и так далее. Но, надеюсь, приведенного выше примера достаточно, чтобы вы продолжили свой путь.

(Предостережение - код не тестировался.)

Ваш код повторно открывает выходной файл и дополняет его каждый раз, когда приходит пакет данных. Очевидно, что лучшим решением является использование LBS.hPutStr для записи в файл с использованием уже открытого дескриптора файла.

То есть вместо:

S.mapM_ (liftIO . (LBS.appendFile "testaws") . LBS.fromStrict) obj

вы хотите использовать:

S.mapM_ (liftIO . (LBS.hPutStr h) . LBS.fromStrict) obj

Конечно, это ссылка на дескриптор h, и откуда это взялось?

Одно из решений — передать его в getObject или иным образом создать перед вызовом тела getObject, например:

getObject cfg bucket key = withFile "output" $ \h -> do
    req <- ...
    ...
    S.mapM_ (liftIO . (LBS.hPutStr h) . LBS.fromStrict) obj
    ...

Или, может быть, вам нужно создать внутри runResourceT... Я не уверен.

Обновление. См. ответ @haoformayor о том, как заставить ResourceT управлять дескриптором файла для вас.