How do I broadcast a message to every open connection, like a chat server, when I cannot keep a list of connections?

← Error Handling and Exceptions · Ref: Q1901

Keep a registry of the connections' VIEWS, not of the connections. A closeable is never a type argument (E81061) - 'List of TCPConnection' and 'Dict of (String, TCPConnection)' are refused - because a collection is not an owner that closes what it holds: every connection already has an owner, the server, which closes it when its handler returns. A TCP handler is given each connection's 'input' and 'output' as a StringInput and a StringOutput, which have no 'operator close', so those may be kept, shared and written to by anyone:

1. On entry, the handler REGISTERS its output view under the member's name.
2. Anyone may BROADCAST: write a line to every registered view.
3. On exit - in a finally, so it happens however the handler ends - the handler DEREGISTERS its view, before the server closes the connection behind it.

The handlers run concurrently, so the registry is shared state: every change goes through a MutexLock (E08251 otherwise), and a MutexLock is made only as a field initialiser (E08256). Here a Room holds 'MutexLock of Roster', and each of its methods enters the lock with a MutexKey function.

This is a recorded design decision, not a gap: EK9 code holds no collections of closeables. Code that must hold many resources at once - a connection pool - gets a Java-backed standard library type that LENDS a resource for one call (the loan pattern, Q1740) and closes them all itself; EK9 code never sees the collection.

In the program below the server's own client connects as alice and then bob. Each join and leave is broadcast to everyone still present, and each client prints what it hears.

Example

defines module qa.errorhandling.broadcasttoconnections

  defines class

    Roster
      members as Dict of (String, StringOutput)?

      Roster()
        members: Dict() of (String, StringOutput)

      add()
        ->
          name as String
          output as StringOutput
        members += DictEntry(name, output)

      remove()
        -> name as String
        members -= name

      tell()
        -> text as String
        for output in members.values()
          output.println(text)

      present() as pure
        <- rtn as Integer: length members

      default operator ?

    Room
      roster as MutexLock of Roster: MutexLock(Roster())

      join()
        ->
          name as String
          output as StringOutput
        joining <- (name, output) is MutexKey of Roster as function
          lockedItem.add(name, output)
          lockedItem.tell(`${name} joined`)
        require roster.enter(joining)

      leave()
        -> name as String
        leaving <- (name) is MutexKey of Roster as function
          lockedItem.remove(name)
          lockedItem.tell(`${name} left, ${lockedItem.present()} present`)
        require roster.enter(leaving)

      default operator ?

  defines function

    connectTo()
      -> port as Integer
      <- rtn as TCPConnection: TCP(NetworkProperties(Networking().loopbackV4(), port)).connect()

  defines program

    BroadcastToConnections()
      stdout <- Stdout()
      port <- 28211
      room <- Room()

      member <- (room) is TCPHandler as function
        name <- input.next()
        room.join(name, output)
        try
          farewell <- input.next()
          require farewell?
        finally
          room.leave(name)
          output.println(`goodbye ${name}`)

      clients <- (port, stdout) is TCPReadyHandler as function
        try
          -> alice <- connectTo(port)
          alice.output().println("alice")
          stdout.println(`alice hears: ${alice.input().next()}`)
          try
            -> bob <- connectTo(port)
            bob.output().println("bob")
            stdout.println(`bob hears: ${bob.input().next()}`)
            stdout.println(`alice hears: ${alice.input().next()}`)
            bob.output().println("bye")
            stdout.println(`bob hears: ${bob.input().next()}`)
            stdout.println(`alice hears: ${alice.input().next()}`)
          alice.output().println("bye")
          stdout.println(`alice hears: ${alice.input().next()}`)

      try
        -> server <- TCP(NetworkProperties(port: port, backlog: 2, maxConcurrent: 2, localOnly: true))
        served <- server.acceptUntilComplete(member, clients)
        stdout.println(`served ${served}`)

Common mistakes

E81061 — A TCPConnection is closeable, and a closeable is never a type argument: a Dict is not an owner that closes what it holds, and each connection's owner already closes it. Register the connection's StringOutput view instead. See ek9 -h E81061 for details.

Incorrect:

      members as Dict of (String, TCPConnection)?

Correct:

      members as Dict of (String, StringOutput)?
Other ways to ask this
  • How do I keep a list of connected clients in EK9?
  • Why can't I hold a List of TCPConnection or a Dict of connections?
  • How do I write a chat server that sends to every client?
  • Can EK9 code keep a pool of connections?

Coming from another language?

Java: a chat server keeps a ConcurrentHashMap of Socket or PrintWriter and must remember to remove a client on every exit path, or it writes to a closed socket - or closes it twice when the map is cleaned up. Go: a map of net.Conn guarded by a mutex, with the same removal discipline. EK9: the registry holds non-closeable StringOutput views behind a MutexLock, the handler deregisters in a finally, and the server alone closes each connection.

Keywords: StringOutput, E81061, TCPHandler, pool, view, MutexLock, connections, chat, broadcast, registry