кабелепровод и розетки: разрешить несколько подключений

Из документации sinkTBMChan :

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

Координация отключения писателей и читателей с канала — нетривиальная проблема, но для ее решения можно повторно использовать решение из экосистемы pipes, то есть использовать библиотеку pipes-concurrency. Эта библиотека предоставляет несколько pipes-независимых утилит, которые вы можете повторно использовать с conduit библиотеками для связи между читателями и писателями, чтобы каждая сторона автоматически правильно знала, когда нужно очистить, и вы также можете вручную очистить любую сторону.

Ключевая функция, которую вы используете из библиотеки pipes-concurrency, называется spawn. Его тип:

spawn :: Buffer a -> IO (Output a, Input a)

Buffer указывает, какую базовую абстракцию канала STM использовать. Судя по вашему примеру кода, похоже, вам нужен буфер Bounded:

spawn (Bounded 8) :: IO (Output a, Input a)

В этом случае a может быть любым, поэтому это может быть ByteString, например:

spawn (Bounded 8) :: IO (Output ByteString, Input ByteString)

Input и Output ведут себя как почтовый ящик. Вы добавляете сообщения в почтовый ящик путем sendпереноса данных в Outputs, а вы извлекаете сообщения из почтового ящика (в порядке FIFO) путем recvпереноса данных из Inputs:

-- Returns `False` if the mailbox is sealed
send :: Output a -> a -> STM Bool

-- Returns `Nothing` if the mailbox is sealed
recv :: Input a -> STM (Maybe a)

Отличительной особенностью pipes-concurrency является то, что он позволяет сборщику мусора автоматически запечатывать почтовый ящик, если в почтовый ящик нет ни читателей, ни писателей. Это позволяет избежать общего источника взаимоблокировок.

Если бы вы использовали экосистему pipes, вы обычно использовали бы следующие две высокоуровневые утилиты для чтения и записи в почтовый ящик.

-- Stream values into the mailbox until it is sealed
toOutput :: Output a -> Consumer a IO ()

-- Stream values from the mailbox until it is sealed
fromInput :: Input a -> Producer a IO ()

Однако, поскольку базовый механизм pipes независим, вы можете переписать эквивалентные conduit версии этих функций:

import Control.Monad.Trans.Class (lift)
import Data.Conduit
import Pipes.Concurrent

toOutput' :: Output a -> Sink a IO ()
toOutput' o = awaitForever (\a -> lift $ atomically $ send o a)

fromInput' :: Input a -> Source IO a
fromInput' i = do
    ma <- lift $ atomically $ recv i
    case ma of
        Nothing -> return ()
        Just a  -> do
            yield a
            fromInput' i

Тогда ваша основная функция будет выглядеть примерно так:

main :: IO ()
main = do
    soc <- socket AF_UNIX Stream 0
    bind soc (SockAddrUnix "mysock")
    (output, input) <- spawn (Bounded 8)
    forkIO $ readFromSocket soc $$ toOutput output
    fromInput input $$ DCB.sinkHandle stdout
  removeFile "mysock"

... где readFromSocket будет некоторым Source, который читается из вашего Socket.

Затем вы можете свободно писать в output, используя также другие источники данных, и не беспокоиться о необходимости их координации или правильной утилизации input или output, когда вы закончите.

Чтобы узнать больше о pipes-concurrency, я рекомендую прочитать официальный учебник.

Я думаю, что ответ @shang правильный, я бы пошел немного дальше и сказал, что поведение writeTBMChan выглядит здесь лучшим виновником. Я бы рекомендовал изменить его, чтобы автоматически не закрывать файл TBMChan. Простая реализация этой идеи:

sinkTBMChan chan = awaitForever $ liftIO . atomically . writeTBMChan chan

Если вы используете это в своей программе, она будет работать так, как ожидалось.

Итак, вот один ответ, который не включает создание возобновляемой раковины. sourceSocket в network-conduit позволяет одно соединение, но мы можем реализовать поведение повторного подключения внутри sourceSocket (извините за код, я думаю, что его нужно почистить, но, по крайней мере, он работает!):

sourceSocket :: (MonadIO m) => Socket -> Producer m ByteString
sourceSocket sock =
    loop
  where
    loop = do
      (conn, _) <- lift . liftIO $ accept sock
      loop' conn
      lift . liftIO $ close conn
      loop
    loop' conn = do
      bs <- lift . liftIO $ recv conn 4096
      if B.null bs
        then return ()
        else yield bs >> loop' conn

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