sockets - 导管和 socket : allow multiple connections

Question

下面是一些使用 conduit 实现小型接收服务器的代码。 , network-conduit , 和 stm-conduit .它在套接字上接收数据，然后通过 STM channel 将其流式传输到主线程。

import Control.Concurrent (forkIO)
import Control.Concurrent.STM (atomically)
import Control.Concurrent.STM.TBMChan (newTBMChan, TBMChan())
import Control.Monad (void)
import Control.Monad.IO.Class (MonadIO (liftIO))
import Control.Monad.Trans.Class

import Data.ByteString (ByteString)
import qualified Data.ByteString as B
import Data.Conduit
import qualified Data.Conduit.Binary as DCB
import Data.Conduit.Extra.Resumable
import Data.Conduit.Network (sourceSocket)
import Data.Conduit.TMChan (sinkTBMChan, sourceTBMChan, mergeSources)

import System.Directory (removeFile)
import System.IO

type BSChan = TBMChan ByteString

listenSocket :: Socket -> Int -> IO BSChan
listenSocket soc bufSize = do
    chan <- atomically $ newTBMChan bufSize
    forkListener chan
    return chan
  where
    forkListener chan = void . forkIO $ listen soc 2 >> loop where 
      loop = do
        (conn, _) <- accept soc
        sourceSocket conn $$ sinkTBMChan chan
        close conn
        loop

main :: IO ()
main = do
  soc <- socket AF_UNIX Stream 0
  bind soc (SockAddrUnix "mysock")
  socChan <- listenSocket soc 8
  sourceTBMChan socChan $$ DCB.sinkHandle stdout
  removeFile "mysock"

(在实际应用程序中，来自套接字的数据流与其他一些数据流合并，这就是为什么我不在监听器中直接处理它的原因)。

问题是，我曾期望它在主线程被杀死之前保持打开状态，而是在套接字上收到第一条消息后退出。我无法弄清楚它为什么这样做，除非它是接收器(在第 2 到最后一行)在看到第一个数据流的结尾时退出。我可以说服它不要这样做吗？ Conduit 中有一些东西关于使源可恢复，但不是接收器。

Answer 1

来自 sinkTBMChan 的文档:

When the sink is closed, the channel will close too.

因此，当第一个套接字句柄关闭时，它会导致 Source来自 sourceSocket关闭，关闭连接的接收器，进而关闭 TBMChan传播到 sinkHandle停止水槽。

解决此问题的最简单方法可能是更改您的 loop到连接之间不关闭的自定义源，并将该源连接到 TBMChan .

listenSocket :: Socket -> Int -> IO BSChan
listenSocket soc bufSize = do
    chan <- atomically $ newTBMChan bufSize
    forkListener chan
    return chan
  where
    forkListener chan = void . forkIO $ do
      listen soc 2
      loop $$ sinkTBMChan chan

    loop = do
      (conn, _) <- liftIO $ accept soc
      sourceSocket conn
      liftIO $ close conn
      loop