An example of how to use `loop` in the purescript-coroutines library for a single producer and consumer, and for multiple consumers connected to a single producer

Connection.purs

module Connection where

import Prelude
import Control.Coroutine (Consumer, Producer, await)
import Control.Coroutine.Aff (produceAff)
import Control.Monad.Aff (Aff, later')
import Control.Monad.Aff.AVar (AVAR)
import Control.Monad.Aff.Console (CONSOLE, log)
import Control.Monad.Eff.Class (liftEff)
import Control.Monad.Eff.Random (RANDOM, randomInt)
import Control.Monad.Rec.Class (forever)
import Control.Monad.Trans.Class (lift)
import Data.Either (Either(..))
import Data.Generic (class Generic, gShow)
import Data.Maybe (Maybe(..))

type PeerID = String
type PeerConnection = String

type Session =
  { peerid :: PeerID
  , connection :: PeerConnection }

data PeerEvent
  = Waiting String
  | Failed String
  | POpen PeerID

data ConnectionEvent
  = COpen
  | CData String
  | CClose

derive instance gPeerEvent :: Generic PeerEvent
derive instance gConnectionEvent :: Generic ConnectionEvent
instance showPeerEvent :: Show PeerEvent where show = gShow
instance showConnectionEvent :: Show ConnectionEvent where show = gShow

-- Simulate a function that connects to a (potentially long-running)
-- service and gets an ID.
getPeerID :: Producer PeerEvent (Aff (avar :: AVAR, console :: CONSOLE, random :: RANDOM)) (Maybe PeerID)
getPeerID = produceAff (\emit -> do

    log "Connecting..."

    later' 1000 $ do
      emit $ Left $ Waiting "Still getting peerid..."

    later' 1000 $ do

      -- Simulate some sort of failure that requires a retry.
      i <- liftEff $ randomInt 0 5
      if i < 4
        then do
          -- If producer was called with `loop`, the
          -- producer will restart.
          emit $ Left $ Failed "Failure to get peerid."
          emit $ Right Nothing
        else emit $ Left $ Waiting "Still getting peerid..."

    later' 2000 $ do
      emit $ Right $ Just $ "id:peerid"

  )

-- Consumer failures events related to getting a PeerID and
-- log them for later diagnostics
logConnectionFailures :: Consumer PeerEvent (Aff (avar :: AVAR, console :: CONSOLE, random :: RANDOM)) PeerID
logConnectionFailures = forever $ do
  event <- await
  lift $ do
    case event of
      (Failed message) -> log $ "LOG: " <> message
      _ -> pure unit

-- Consume events related to getting a PeerID and send them
-- to STDOUT
showConnectionDetails :: Consumer PeerEvent (Aff (avar :: AVAR, console :: CONSOLE, random :: RANDOM)) PeerID
showConnectionDetails = forever $ do
  event <- await
  lift $ do
    log $ "STDOUT: " <> (show event)

output.txt

Running singleConsumerWithProducer
Connecting...
STDOUT: Connection.Waiting "Still getting peerid..."
STDOUT: Connection.Failed "Failure to get peerid."
Connecting...
STDOUT: Connection.Waiting "Still getting peerid..."
singleConsumerWithProducer result: "id:peerid"

Running multipleConsumerWithProducer
Connecting...
STDOUT: Connection.Waiting "Still getting peerid..."
LOG: Failure to get peerid.
STDOUT: Connection.Failed "Failure to get peerid."
Connecting...
STDOUT: Connection.Waiting "Still getting peerid..."
STDOUT: Connection.Waiting "Still getting peerid..."
multipleConsumerWithProducer result: "id:peerid"

TestConnection.purs

module Test.Connection where

import Prelude
import Connection (getPeerID, logConnectionFailures, showConnectionDetails)
import Control.Coroutine (connect, joinConsumers, loop, pullFrom, runProcess, transform, transformConsumer)
import Control.Monad.Aff (Aff, launchAff)
import Control.Monad.Aff.AVar (AVAR)
import Control.Monad.Aff.Console (CONSOLE, log)
import Control.Monad.Eff (Eff)
import Control.Monad.Eff.Exception (EXCEPTION)
import Control.Monad.Eff.Random (RANDOM)
import Control.Monad.Rec.Class (forever)
import Data.Tuple (Tuple(..))


main :: Eff (err :: EXCEPTION, avar :: AVAR, console :: CONSOLE, random :: RANDOM) Unit
main = void $ launchAff $ do
  log "Running singleConsumerWithProducer"
  singleConsumerWithProducer

  log "Running multipleConsumerWithProducer"
  multipleConsumerWithProducer

singleConsumerWithProducer :: Aff (avar :: AVAR, console :: CONSOLE, random :: RANDOM) Unit
singleConsumerWithProducer = do
  v <- runProcess $ connect (loop getPeerID) showConnectionDetails
  log $ "singleConsumerWithProducer result: " <> (show v)


multipleConsumerWithProducer :: Aff (avar :: AVAR, console :: CONSOLE, random :: RANDOM) Unit
multipleConsumerWithProducer = do
  v <- runProcess $
    transformConsumer
      (forever $ transform (\a -> Tuple a a))
      (joinConsumers logConnectionFailures showConnectionDetails)
    `pullFrom` (loop getPeerID)
  log $ "multipleConsumerWithProducer result: " <> (show v)