UDP with Scala and FS2 Explained

The UDP Server

To implement UDP in NIO we’ll make use of the DatagramChannel class which supports concurrent reading and writing, however at most one thread may be reading and at most one thread may be writing at any given time. Create a NioUdp.scala file in the following path, src/main/scala/com/rockthejvm/nioUdp/NioUdp.scala, and add the following code:

package com.rockthejvm.nioUdp
import cats.effect.IOApp

object NioUdp extends IOApp {
  def server: Unit = {
      val port: Int = 5555
      val ip: String = "127.0.0.1"
  }
  def run(args: List[String]): IO[ExitCode] = ???
}

Here we define an IOApp with its run method, we also have our server function where we’ve defined the port number and ip address on which the server will be waiting for incoming connections.

...
import scala.util.Try
import scala.util.Failure
import scala.util.Success
import java.nio.channels.DatagramChannel
import java.net.StandardProtocolFamily

object NioUdp extends IOApp {
  def server: Unit = {
    ...
    Try {
      val datagramChannel: DatagramChannel =
        DatagramChannel.open(StandardProtocolFamily.INET)
      if (datagramChannel.isOpen()) {
        println("[server] Udp server is successfully opened")
      } else {
          println("[server] The channel cannot be opened!")
        }
    } match {
      case Failure(ex) =>
        println(s"[client] ${ex.getMessage}")
      case Success(_) =>
        println("[client] Everything works fine")
    }
  }
  ...
}

We start by opening a Datagram channel using the open() method on DatagramChanel and passing it a ProtocolFamily parameter, here we use StandardProtocolFamily.INET meant for IPv4, the StandardProtocolFamily enum also provides two other options, StandardProtocolFamily.INET6 for IPv6 and StandardProtocolFamily.UNIX for UNIX domain interprocess communication depending on your needs.

We confirm if datagramChannel is indeed open by calling datagramChannel.isOpen() with the help of an if statement whose body will have a notification message printed if this is true or false. This implementation is wrapped in a Try which we pattern match to either print an error message in case of a Failure or a success message in case no errors occur.

Let’s build on our if statement:

...
import java.net.StandardSocketOptions
import java.net.InetSocketAddress

if (datagramChannel.isOpen()) {
  println("[server] Udp server is successfully opened")

  datagramChannel.setOption(StandardSocketOptions.SO_RCVBUF, 4 * 1024)
  datagramChannel.setOption(StandardSocketOptions.SO_SNDBUF, 4 * 1024)

  datagramChannel.bind(new InetSocketAddress(ip, port))
  println(
    "[server] Udp server is bound to:" + datagramChannel.getLocalAddress()
  )
}
...

Here we define some socket options using the setOption() method which takes a SocketOption and it’s associated value.

The StandardSocketOptions.SO_RCVBUF and StandardSocketOptions.SO_SNDBUF are options for the datagram socket receive buffer and socket send buffer respectively which we set to 4 kilobytes each.

The size could be increased for high-volume connections and in the case of the receive buffer, reduced to limit the possible backlog of incoming data. StandardSocketOptions provides multiple options for fine-grained control of a DatagramChannel:

We also make sure to bind our datagramChannel to a SocketAddress using the bind() method, this creates an association between the socket and the local address. In the code above we use InetSocketAddress which is a subclass of SocketAddress, and in the next line, print the address it’s bound to.

Let’s see how we can send our data packets:

...
import java.nio.ByteBuffer
import java.net.SocketAddress

val content: ByteBuffer = ByteBuffer.allocate(65507)
if (datagramChannel.isOpen()) {
  ...
  while (true) {
    val clientAddress: SocketAddress = datagramChannel.receive(content)
    content.flip()
    println(
      s"[server] I've received ${content.limit()} bytes " +
        s"from ${clientAddress.toString()}! Sending them back ..."
    )
    datagramChannel.send(content, clientAddress)
    content.clear()
  }
}
...

Here we showcase how the echo server works. The logic is defined within an infinite while loop where our datagram channel continuously listens for incoming Datagrams or packets from clients.

We use the datagramChannel.receive() method to capture these packets, it takes a ByteBuffer as a parameter which we defined as content, this method transfers the incoming datagrams to the ByteBuffer like a write operation. This operation returns the SocketAddress of the client.

The flip() method on content makes the ByteBuffer ready for a new sequence of channel writes. We then print a message showing the bytes received and the client address, then we send content which now contains what was sent by the client back by calling datagramChannel.send(), this takes a ByteBuffer and SocketAddress. Finally, we call clear() on content to reuse the buffer.

Here’s the full code:

package com.rockthejvm.nioUdp

import cats.effect.{IOApp, ExitCode, IO}
import scala.util.Try
import scala.util.Failure
import scala.util.Success
import java.nio.channels.DatagramChannel
import java.net.StandardProtocolFamily
import java.net.StandardSocketOptions
import java.net.InetSocketAddress
import java.nio.ByteBuffer
import java.net.SocketAddress

object NioUdp extends IOApp {
  def server: Unit = {
    val port: Int = 5555
    val ip: String = "127.0.0.1"
    val content: ByteBuffer = ByteBuffer.allocate(65507)

    Try {
      val datagramChannel: DatagramChannel =
        DatagramChannel.open(StandardProtocolFamily.INET)
      if (datagramChannel.isOpen()) {
        println("[server] Udp server is successfully opened")

        datagramChannel.setOption(StandardSocketOptions.SO_RCVBUF, 4 * 1024)
        datagramChannel.setOption(StandardSocketOptions.SO_SNDBUF, 4 * 1024)

        datagramChannel.bind(new InetSocketAddress(ip, port))
        println(
          "[server] Udp server is bound to:" + datagramChannel.getLocalAddress()
        )

        while (true) {
          val clientAddress: SocketAddress = datagramChannel.receive(content)
          content.flip()
          println(
            s"[server] I've received ${content.limit()} bytes " +
              s"from ${clientAddress.toString()}! Sending them back ..."
          )
          datagramChannel.send(content, clientAddress)
          content.clear()
        }
      } else {
        println("[server] The channel cannot be opened!")
      }
    } match {
      case Failure(ex) =>
        println(s"[server] ${ex.getMessage}")
      case Success(_) =>
        println("[server] Everything works fine")
    }
  }

    def run(args: List[String]): IO[ExitCode] = ???
}

The UDP Client

Creating a UDP client is very similar to creating a server, the only change is we’ll be preparing the data to send to the server which we later decode and print once it’s echoed back. Here’s how it’s implemented:

import java.nio.charset.Charset
import java.nio.charset.CharsetDecoder

object NioUdp extends IOApp {
  ...
  def client = {
    val serverPort: Int = 5555
    val serverIp: String = "127.0.0.1"

    val charSet: Charset = Charset.defaultCharset()
    val decoder: CharsetDecoder = charSet.newDecoder()

    val echoText: ByteBuffer =
      ByteBuffer.wrap("[message] I was sent back from the server!".getBytes())
    val content: ByteBuffer = ByteBuffer.allocate(65507)

    Try {
      val datagramChannel = DatagramChannel.open(StandardProtocolFamily.INET)
      if (datagramChannel.isOpen()) {
        datagramChannel.setOption(StandardSocketOptions.SO_RCVBUF, 4 * 1024)
        datagramChannel.setOption(StandardSocketOptions.SO_SNDBUF, 4 * 1024)

        val sent = datagramChannel.send(
          echoText,
          new InetSocketAddress(serverIp, serverPort)
        )
        println(s"[client] I have successfully sent $sent bytes to the Echo Server!")

        datagramChannel.receive(content)

        content.flip()
        val charBuffer = decoder.decode(content)
        println(charBuffer.toString())
        content.clear()
        datagramChannel.close()
      } else {
        println("[client] The channel cannot be opened!")
      }
    } match {
      case Failure(ex) =>
        println(s"[client] ${ex.getMessage}")
      case Success(_) =>
        println("[client] Everything works fine")
    }
  }
  ...
}

Here we define a Charset by calling Charset.defaultCharset(), then create a decoder using the newDecoder() method on CharSet which will be used to decode the byte data sent by the server. We also define echoText, a ByteBuffer containing the text that we want to send, we use the ByteBuffer.wrap() method that wraps our message into the buffer.

We also define a DatagramChannel, check if it’s open then send echoText to the server using the send() method. When we receive the response from the server, it’s written to content and decoded using the decode() method from the CharsetDecoder that we defined earlier. We then close the socket by calling datagramChannel.close() since we are done sending messages.

Lastly let’s define our run method and then test the application:

import cats.syntax.parallel.*
import scala.concurrent.duration.*

object NioUdp extends IOApp {
  ...
  def run(args: List[String]): IO[ExitCode] =
    (IO(server), IO.sleep(500.millis).flatMap(_ => IO(client)))
      .parMapN((s, c) => ())
      .as(ExitCode.Success)
}

Here we wrap both functions in IO and then run them concurrently using the parMapN() from cats, however, we delay the execution of the client by 500 milliseconds to give the server time to start. These are the results after running the application

[server] Udp server is successfully opened
[server] Udp server is bound to:/127.0.0.1:5555
[client] I have successfully sent 42 bytes to the Echo Server!
[server] I've received 42 bytes from /127.0.0.1:49841! Sending them back ...
[message] I was sent back from the server!
[client] Everything works fine

Looking at the output everything works, notice that we didn’t bind an address and port number to the client, this was done automatically by NIO, if we run this program again, a different address will be assigned.

A Connected Client

NIO provides an alternative client implementation where the client first connects to the server before sending and receiving packets, there are some minor changes compared to the previous section, here’s the code:

  def connectedClient = {
    val serverPort: Int = 5555
    val serverIp: String = "127.0.0.1"

    val charSet: Charset = Charset.defaultCharset()
    val decoder: CharsetDecoder = charSet.newDecoder()

    val echoText: ByteBuffer =
      ByteBuffer.wrap("[message] I was sent back from the server!".getBytes())
    val content: ByteBuffer = ByteBuffer.allocate(65507)

    Try {
      val datagramChannel = DatagramChannel.open(StandardProtocolFamily.INET)
      if (datagramChannel.isOpen()) {
        datagramChannel.setOption(StandardSocketOptions.SO_RCVBUF, 4 * 1024)
        datagramChannel.setOption(StandardSocketOptions.SO_SNDBUF, 4 * 1024)

        datagramChannel.connect(new InetSocketAddress(serverIp, serverPort))

        if(datagramChannel.isConnected()){
          val sent = datagramChannel.write(echoText)
          println(
            s"[client] I have successfully sent $sent bytes to the Echo Server!"
          )

          datagramChannel.read(content)

          content.flip()
          val charBuffer = decoder.decode(content)
          println(charBuffer.toString())
          content.clear()
          datagramChannel.close()
        }
      } else {
        println("[client] The channel cannot be opened!")
      }
    } match {
      case Failure(ex) =>
        println(s"[client] ${ex.getMessage}")
      case Success(_) =>
        println("[client] Everything works fine")
    }
  }

After the datagramChannel is open, we connect to the server by calling datagramChannel.connect() and passing it an InetSocketAddress, this is the main difference. We also check if the connection is established by calling datagramChannel.isConnected(), then implement our code.

Another change is the use datagramChannel.write and datagramChannel.read to send and receive our messages. In addition, we no longer need to pass the InetSocketAddress to the write() since the datagram channel is already connected to the server address and knows where the data will be sent.

Why use this method? Once datagramChannel.connect() is called, the read() and write() methods will only send and receive datagram packets from the address it’s connected to.

Lastly, we can either call datagramChannel.disconnect() or datagramChanel.close() once we are done using the channel. Calling datagramChannel.disconnect() on a closed channel or one that’s not connected has no effect. If we call our run method with connectedClient, the results should be the same.

Let’s recap on how we implemented UDP:

• Defined the port and address values.
• Opened a DatagramChannel and set the ProtocolFamily value.
• Set some Socket options.
• Bind the socket to an address in case of a server.
• Then sent and received the data.
• Lastly, close the socket in case of a client.

Can we do the same with a higher-level API like FS2?

The UDP Server

Create Fs2Udp.scala in the following path, src/main/scala/com/rockthejvm/fs2Udp/Fs2Udp.scala and add the following code:

package com.rockthejvm.fs2Udp

import cats.effect.{IOApp, ExitCode, IO}
import fs2.Stream
import fs2.io.net.Network
import com.comcast.ip4s.*

object Fs2Udp extends IOApp {
  def server =
    Stream
      .resource(
        Network[IO].openDatagramSocket(
          address = Some(ip"127.0.0.1"),
          port = Some(port"5555"),
        )
      )

  def run(args: List[String]): IO[ExitCode] = ???
}

Just like Java NIO, we start by opening a DatagramChannel, or in this case a DatagramSocket using the Network[IO].openDatagramSocket() method and passing it address and port values.

The com.comcast.ip4s library provides us with convenient string interpolation methods for constructing these values. Wrapping this method in Stream.resource() gives us a Stream[IO, DatagramSocket[IO]].

Let’s set the socket options, and ProtocolFamily :

...
import fs2.io.net.SocketOption
import java.net.StandardSocketOptions
import java.net.StandardProtocolFamily

object Fs2Udp extends IOApp {
  def server =
    Stream
      .resource(
        Network[IO].openDatagramSocket(
          address = Some(ip"127.0.0.1"),
          port = Some(port"5555"),
          options = List(
            SocketOption(StandardSocketOptions.SO_RCVBUF, 1024 * 4),
            SocketOption(StandardSocketOptions.SO_SNDBUF, 1024 * 4)
          ),
          protocolFamily = Some(StandardProtocolFamily.INET)
        )
      )

    def run(args: List[String]): IO[ExitCode] = ???
}

To define socket options, the openDatagramSocket() method allows for an options parameter that takes a List[DatagramSocketOption], if we dig into this type definition, we’ll find that DatagramSocketOption is an alias for the SocketOption trait from fs2.io.net:

package object net {
...
  type DatagramSocketOption = SocketOption
  val DatagramSocketOption = SocketOption
}

The SocketOption companion object has an apply method that takes a StandardSocketOptions option as a key with its corresponding value, the same as Java NIO. We also define the protocol family using StandardProtocolFamily.INET for IPv4.

It’s also fine to provide only the port value to openDatagramSocket():

object Fs2Udp extends IOApp {
  def server =
    Stream
      .resource(
        Network[IO].openDatagramSocket(
          port = Some(port"5555")
        )
      )
  ...
}

This still works since 127.0.0.1 also known as Localhost is the address IP address of the local computer, and allows the machine to connect and communicate with itself, it’s set by default. The protocol family and StandardSocketOptions, SO_RCVBUF, and SO_SNDBUF also have default values depending on your system.

def server =
  Stream
    .resource(
      Network[IO].openDatagramSocket(
        ...
      )
    )
    .evalTap(socket =>
      IO.println("[server] Udp server is successfully opened") *>
        socket.localAddress.flatMap { address =>
          IO.println(s"[server] Udp server is bound to: ${address.toString}")
        }
    )

Here we use the evalTap() function on fs2.Stream[IO, DatagramSocket[IO]] to print out some observations, this gives us access to the DatagramSocket[IO] but still keeps it available for the next operation.

If the openDatagramSocket() method succeeds, we print a successfully opened message as well as the address that the server is bound to by flat mapping on socket.localAddress and printing the address of type SocketAddress[IpAddress] as a string.

The next step is to read incoming datagrams:

def server =
  Stream
    .resource(
      ...
    )
    .evalTap(socket =>
      ...
    )
    .flatMap { socket =>
      socket.reads
        .evalTap { datagram =>
          IO.println(
            s"[server] I've received ${datagram.bytes.size} bytes " +
              s"from ${datagram.remote.toString}! Sending them back ..."
          )
        }
        .through(socket.writes)
    }

The reads() method on DatagramSocket[IO] returns a Stream[IO, Datagram], we evalTap on this to print out the bytes size of the incoming datagram and the address of the client by calling datagram.bytes.size and datagram.remote.toString respectively.

Lastly, calling through(socket.writes) takes the Stream of incoming Datagram’s and writes it back to the address where they originated:

The final step is to handle any errors that may occur using handleErrorWith:

def server =
  Stream
    .resource(
      ...
    )
    .evalTap(socket =>
      ...
    )
    .flatMap { socket =>
      ...
    }
    .handleErrorWith { error =>
        Stream.eval(
          IO.println(s"[server] ${error.getMessage}")
        )
      }

Here we print the error message as a string by calling error.getMessage.

This is the full code:

package com.rockthejvm.fs2Udp

import cats.effect.{IOApp, ExitCode, IO}
import fs2.Stream
import fs2.io.net.Network
import com.comcast.ip4s.*
import fs2.io.net.SocketOption
import java.net.StandardSocketOptions
import java.net.StandardProtocolFamily
import fs2.io.net.DatagramSocket

object Fs2Udp extends IOApp {
  def server =
    Stream
      .resource(
        Network[IO].openDatagramSocket(
          address = Some(ip"127.0.0.1"),
          port = Some(port"5555"),
          options = List(
            SocketOption(StandardSocketOptions.SO_RCVBUF, 1024 * 4),
            SocketOption(StandardSocketOptions.SO_SNDBUF, 1024 * 4)
          ),
          protocolFamily = Some(StandardProtocolFamily.INET)
        )
      )
      .evalTap(socket =>
        IO.println("[server] Udp server is successfully opened") *>
          socket.localAddress.flatMap { address =>
            IO.println(s"[server] Udp server is bound to: ${address.toString}")
          }
      )
      .flatMap { socket =>
        socket.reads
          .evalTap { datagram =>
            IO.println(
              s"[server] I've received ${datagram.bytes.size} bytes " +
                s"from ${datagram.remote.toString}! Sending them back ..."
            )
          }
          .through(socket.writes)
      }
      .handleErrorWith { error =>
        Stream.eval(
          IO.println(s"[server] ${error.getMessage}")
        )
      }
  def run(args: List[String]): IO[ExitCode] = ???
}

The UDP Client

The UDP client will look very similar to our server with a few minor changes. Let’s define the client function:

object Fs2Udp extends IOApp {
  ...
  def client = {
    val serverAddress: SocketAddress[IpAddress] =
      SocketAddress(ip"127.0.0.1", port"5555")
    Stream
      .resource(Network[IO].openDatagramSocket())
  }
  ...
}

We start by defining the socket address of the server as a SocketAddress[IpAddress], this is a case class that comes from com.comcast.ip4s and should not be confused with SocketAddress from java.net.

Next, we open the datagram socket but don’t supply any arguments since we’ll be assigned by the system, server ports need to be known beforehand, however, client ports can be ephemeral meaning they are discarded after use.

Our server will get the client address from the datagram header and use that to echo back.

import fs2.text

def client = {
    ...
    Stream
      .resource(...)
      .flatMap { socket =>
        Stream("[message] I was sent back from the server!")
          .through(text.utf8.encode)
          .chunks
          .flatMap { data =>
            Stream(Datagram(serverAddress, data))
              .evalTap { datagram =>
                IO.println(
                  s"[client] Sending ${datagram.bytes.size} bytes to the Echo Server!"
                )
              }
          }.through(socket.writes)
          .drain
      }
}

In this section we prepare our message as an fs2.Stream, encode it into a stream of bytes by calling through(text.utf8.encode) and outputting chunks from the resulting stream.

We then flatMap and pass data of type Chunk[Byte] to a Datagram stream along with serverAddress and print out the size of the data being sent. Lastly, we write this data the server and drain the stream.

...
.through(socket.writes)
.drain ++
socket.reads
  .flatMap(datagram => Stream.chunk(datagram.bytes))
  .through(text.utf8.decode)
  .foreach { response =>
    IO.println(response)
  }

Here we use the ++ operator to concatenate another stream that reads datagrams from the same socket resulting in a Stream[IO, Datagram] whose bytes we then chunk, decode, and print to the console. Notice that the process here is the reverse of the previous paragraph.

Because of the ++ operator, socket.reads is only starts after the previous stream has completed.

Here’s the full code:

import fs2.text

def client = {
    val serverAddress: SocketAddress[IpAddress] =
      SocketAddress(ip"127.0.0.1", port"5555")
    Stream
      .resource(Network[IO].openDatagramSocket())
      .flatMap { socket =>
        Stream("[message] I was sent back from the server!")
          .through(text.utf8.encode)
          .chunks
          .flatMap { data =>
            Stream(Datagram(serverAddress, data))
              .evalTap { datagram =>
                IO.println(
                  s"[client] Sending ${datagram.bytes.size} bytes to the Echo Server!"
                )
              }
          }
          .through(socket.writes)
          .drain ++
          socket.reads
            .flatMap(datagram => Stream.chunk(datagram.bytes))
            .through(text.utf8.decode)
            .foreach { response =>
              IO.println(response)
            }
      }
      .handleErrorWith { error =>
        Stream.eval(
          IO.println(s"[client] ${error.getMessage}")
        )
      }
  }

We also make sure to handle any errors that may occur by calling handleErrorWith.

Let’s define the run method and test our application:

...
import scala.concurrent.duration.*

object Fs2Udp extends IOApp {
  ...
  def run(args: List[String]): IO[ExitCode] =
    server
      .concurrently(Stream.sleep[IO](500.millis) ++ client)
      .compile
      .drain
      .as(ExitCode.Success)
}

Just like before we run both server and client concurrently but give the server 500 milliseconds to first start up. Here are the results:

[server] Udp server is successfully opened
[server] Udp server is bound to: 127.0.0.1:5555
[client] Sending 42 bytes to the Echo Server!
[server] I've received 42 bytes from 127.0.0.1:49763! Sending them back ...
[message] I was sent back from the server!

The fs2.io.net NIO abstractions use the DatagramChannel’s send() and receive() methods under the hood since it’s possible to send and receive datagrams to and from different addresses.

Multicast Server in NIO

For this example, we’ll create a server that sends the current date and time as an endless stream to any client that connects to the multicast group. Create a file called NioUdpMulticast.scala in the following path, src/main/scala/com/rockthejvm/nioUdp/NioUdpMulticast.scala and add the following code:

package com.rockthejvm.nioUdp

import cats.effect.{IOApp, ExitCode, IO}
import java.nio.ByteBuffer
import scala.util.{Try, Failure, Success}
import java.nio.channels.DatagramChannel
import java.net.{
  StandardProtocolFamily,
  StandardSocketOptions,
  InetSocketAddress,
  NetworkInterface,
  InetAddress
}
import java.util.Date

object NioUdpMulticast extends IOApp {
  def server: Unit = {
    val port: Int = 5555
    val groupIp: String = "225.4.5.6"
    val content: ByteBuffer = ByteBuffer.allocate(65507)

    Try {
      val datagramChannel: DatagramChannel =
        DatagramChannel.open(StandardProtocolFamily.INET)

      if (datagramChannel.isOpen()) {
        println("[multicast server] Udp server is successfully opened")

        val networkInterface: NetworkInterface =
          NetworkInterface.getByName(
            "wlxb4b024bc35a7"
          ) // Pass the interface name that matches what you have.

        datagramChannel.setOption(
          StandardSocketOptions.IP_MULTICAST_IF,
          networkInterface
        )
        datagramChannel.setOption(StandardSocketOptions.SO_REUSEADDR, true)

        datagramChannel.bind(new InetSocketAddress(port))
        println(
          s"[multicast server] Udp server is bound to: ${datagramChannel.getLocalAddress()}"
        )

        println(
          "[multicast server] Udp server will start sending date time info shortly..."
        )

        while (true) {
          try {
            Thread.sleep(10000)
          } catch {
            case e => println(e.getMessage)
          }

          println("Sending data ...")
          val datetime = ByteBuffer.wrap(new Date().toString().getBytes())
          datagramChannel.send(
            datetime,
            new InetSocketAddress(InetAddress.getByName(groupIp), port)
          )
          datetime.flip()
        }
      } else {
        println("[multicast server] The channel cannot be opened!")
      }
    } match {
      case Failure(ex) =>
        println(s"[multicast server] ${ex.getMessage}")
      case Success(_) =>
        println("[multicast server] Everything works fine")
    }
  }

  def run(args: List[String]): IO[ExitCode] = ???
}

Let’s go through and compare the changes to the previous NIO server implementation:

First, we changed the IP address from localhost to a Multicast address, 225.4.5.6. This address will form the multicast group and is where the datagrams will be sent.

We also added two new socketOptions, StandardSocketOptions.IP_MULTICAST_IF and StandardSocketOptions.SO_REUSEADDR.

The IP_MULTICAST_IF option is used to set the network interface that will be used for sending multicast datagrams, here I’ll be using my wifi network adapter by calling NetworkInterface.getByName("wlxb4b024bc35a7") and passing the interface display name. Make sure you use the name of the multicast interface that you have on your computer or laptop.

The SO_REUSEADDR option is used to tell the system if the address can be reused, this enables multiple programs to bind to the same address and is essential for multicasting.

When calling the bind method, we only provide the port number, the InetSocketAddress has a constructor that allows for this, meaning that it provides a wildcard or placeholder address, typically 0.0.0.0.

Lastly, we have an infinite while loop that sends a date string every 10,000 milliseconds to all members of the multicast group by calling datagramChannel.send() and passing it the date, and the multicast socket address got by calling new InetSocketAddress(InetAddress.getByName(groupIp), port).

Multicast Client in NIO

Let’s start by creating our client function and add the following code:

...
import java.nio.charset.Charset
import java.nio.charset.CharsetDecoder
import java.nio.channels.MembershipKey

object NioUdpMulticast extends IOApp {
  ...
  def client = {
    val port: Int = 5555
    val groupIp: String = "225.4.5.6"

    val charSet: Charset = Charset.defaultCharset()
    val decoder: CharsetDecoder = charSet.newDecoder()
    val content: ByteBuffer = ByteBuffer.allocate(65507)

    Try {
      val datagramChannel = DatagramChannel.open(StandardProtocolFamily.INET)

      val group: InetAddress = InetAddress.getByName(groupIp)
      if (group.isMulticastAddress()) {
        if (datagramChannel.isOpen()) {
          val networkInterface = NetworkInterface.getByName("wlxb4b024bc35a7")
          datagramChannel.setOption(
            StandardSocketOptions.SO_REUSEADDR,
            true
          )

          datagramChannel.bind(new InetSocketAddress(port))
          println(
            s"[multicast client] client is bound to: ${datagramChannel.getLocalAddress()}"
          )

          val key: MembershipKey = datagramChannel.join(group, networkInterface)
          while (true) {
            if (key.isValid()) {
              datagramChannel.receive(content)
              content.flip()
              val charBuffer = decoder.decode(content)
              println(charBuffer.toString())
              content.clear()
            } else {
              println("[multicast client] Invalid join key")
            }
          }
        } else {
          println("[multicast client] The channel cannot be opened!")
        }
      } else {
        println("[multicast client] This is not multicast address")
      }
    } match {
      case Failure(ex) =>
        println(s"[multicast client] ${ex.getMessage}")
      case Success(_) =>
        println("[multicast client] Everything works fine")
    }
  }
  ...
}

Again, the multicast client looks similar to the previous NIO client with a few changes:

Similar to the multicast server, we start by defining the port number and multicast group IP address, groupIp.

Within our Try we first define group as an InetAddress by calling InetAddress.getByName(groupIp), this gives us access to some extra methods on group, one of which is group.isMulticastAddress() that checks if the address provided is an IP multicast address. Once this is confirmed, within an if statement we proceed to open the datagramChannel and set the SO_REUSEADDR socket option to true.

Unlike the previous NIO client implementations, for multicast we bind the datagramChannel to a SocketAddress, here we use a placeholder address and port 5555 similar to our server.

The next step is to join a multicast group, we do this by calling datagramChannel.join(group, networkInterface) where the join() method takes our multicast address and a network interface. This returns a MembershipKey which is a token representing the membership of the IP multicast group.

Within the while loop, we first check if the MembershipKey is valid then start receiving datagrams. MembershipKey also has a drop() method which can be called if we want to opt out of the multicast group.

Lastly, let’s define our run method and test our application:

import cats.syntax.parallel.*

object NioUdpMulticast extends IOApp {
  ...
  def run(args: List[String]): IO[ExitCode] =
    (IO(server), IO(client)).parMapN((s, c) => ()).as(ExitCode.Success)
}

Here are the results:

[multicast server] Udp server is successfully opened
[multicast server] Udp server is bound to: /0.0.0.0:5555
[multicast server] Udp server will start sending date time info shortly...
[multicast client] client is bound to: /0.0.0.0:5555
Sending data ...
Tue Oct 24 13:48:19 EAT 2023
Sending data ...
Tue Oct 24 13:48:29 EAT 2023
Sending data ...
Tue Oct 24 13:48:39 EAT 2023
Sending data ...
Tue Oct 24 13:48:49 EAT 2023

When we analyse the output we can see that both the multicast server and client are bound to the same port, this is possible because we set the SO_REUSEADDR to true in both functions.

Here’s a recap of how UDP multicasting works:

• Defined the port and multicast group address values.
• Check if the address is indeed multicast.
• Open a DatagramChannel and set the ProtocolFamily value.
• Set the following Socket options, IP_MULTICAST_IF for the multicast server and SO_REUSEADDR for both the multicast server and client.
• Bind the socket to an address.
• Join a multicast group in case of a client.
• Start receiving data sent by the multicast server
• Opt out of membership if needed.

Let’s see if we can translate this to FS2.

Multicast Server in FS2

Let’s create a new file, Fs2UdpMulticast.scala in the following path, src/main/scala/com/rockthejvm/fs2Udp/Fs2UdpMulticast.scala and add the following server code:

package com.rockthejvm.fs2Udp

import cats.effect.{IOApp, ExitCode, IO}
import com.comcast.ip4s.*
import fs2.{Stream, text}
import fs2.io.net.Network
import fs2.io.net.SocketOption
import java.net.StandardSocketOptions
import java.net.StandardProtocolFamily
import java.net.NetworkInterface
import java.util.Date
import fs2.io.net.Datagram
import scala.concurrent.duration.*

object Fs2UdpMulticasting extends IOApp {
  def server = {
    val multicastAddress = SocketAddress(ip"225.4.5.6", port"5555")
    Stream
      .resource(
        Network[IO].openDatagramSocket(
          port = Some(port"5555"),
          options = List(
            SocketOption(
              StandardSocketOptions.IP_MULTICAST_IF,
              NetworkInterface.getByName("wlxb4b024bc35a7")
            ),
            SocketOption(
              StandardSocketOptions.SO_REUSEADDR,
              true
            )
          ),
          protocolFamily = Some(StandardProtocolFamily.INET)
        )
      )
      .evalTap{socket =>
        IO.println("[multicast server] Udp server is successfully opened") *>
          socket.localAddress.flatMap(addr =>
            IO.println(
              s"[multicast server] Udp server is bound to: ${addr.toString}"
            )
          ) *>
          IO.println(
            "[multicast server] Udp server will start sending date time info shortly..."
          )
      }
      .flatMap { socket =>
        Stream
          .repeatEval(IO(new Date().toString()))
          .through(text.utf8.encode)
          .chunks
          .map(data => Datagram(multicastAddress, data))
          .metered(10000.millis)
          .evalTap(_ => IO.println("Sending data ..."))
          .through(socket.writes)
          .drain
      }
      .handleErrorWith { error =>
        Stream.eval(
          IO.println(s"[multicast server]: ${error.getMessage}")
        )
      }
  }

  def run(args: List[String]): IO[ExitCode] = ???
}

Most of this should look familiar by now, we start by opening the UDP socket by calling Network[IO].openDatagramSocket(), we pass it the port value but leave the server address value to default to a placeholder address.

We then add IP_MULTICAST_IF and SO_REUSEADDR socket options, and pass StandardProtocolFamily.INET as the protocolFamily value.

Next, we call evalTap() on our resource to give us access to the socket, then use this to print the address that the server is bound to by flat mapping on socket.localAddress along with other relevant messages.

In the following section, we call flatMap and then create an infinite stream of date values by calling Stream.repeatEval(IO(new Date().toString())), these values then are encoded, chunked, and mapped to a Datagram along with the multicastAddress.

Before we write our values, the stream is metered so that any client that joins this multicast group will receive a date value every 10,000 milliseconds.

We end by handling any errors with handleErrorWith().

Multicast Client in FS2

Append the following client code still within the Fs2UdpMulticasting object:

object Fs2UdpMulticasting extends IOApp {
  ...
  def client = {
    Stream
      .resource(
        Network[IO].openDatagramSocket(
          port = Some(port"5555"),
          options = List(
            SocketOption(
              StandardSocketOptions.SO_REUSEADDR,
              true
            )
          ),
          protocolFamily = Some(StandardProtocolFamily.INET)
        )
      )
      .evalTap { socket =>
        socket.localAddress.flatMap(addr =>
          IO.println(
            s"[multicast server] client is bound to: ${addr.toString}"
          )
        )
      }
      .flatMap { socket =>
        val groupMembership: IO[socket.GroupMembership] = socket.join(
          MulticastJoin.fromString("225.4.5.6").get,
          NetworkInterface.getByName("wlxb4b024bc35a7")
        )

        Stream.eval(groupMembership) ++
          socket.reads
            .flatMap(datagram => Stream.chunk(datagram.bytes))
            .through(text.utf8.decode)
            .foreach { response =>
              IO.println(response)
            }
      }
      .handleErrorWith { error =>
        Stream.eval(
          IO.println(s"[multicast client] Error: ${error.getMessage}")
        )
      }
  }
  ...
}

Similar to our server, we open the datagram, set the port value, set SO_REUSEADDR as a socket option, add a protocolFamily and evalTap to print the socket address.

Next we flatMap, and call the join() method on socket to join a multicast group. join() takes a MulticastJoin[IpAddress] from com.comcast.ip4s and a NetworkInterface to return an IO[socket.GroupMembership] that we assigned to groupMembership.

We then call Stream.eval(groupMembership) and join this Stream to socket.reads using ++, lastly we use the handleErrorWith() to print out any error that may occur.

The GroupMembership trait has a drop() method we can call in case we need to leave the multicast group.

Let’s create our run() method and test the application:

object Fs2UdpMulticasting extends IOApp {
  ...
  def run(args: List[String]): IO[ExitCode] =
    server.concurrently(client).compile.drain.as(ExitCode.Success)
}

Here are the results:

[multicast server] Udp server is successfully opened
[multicast server] Udp server is bound to: 0.0.0.0:5555
[multicast server] Udp server will start sending date time info shortly...
[multicast server] client is bound to: 0.0.0.0:5555
Sending data ...
Tue Oct 24 15:21:50 EAT 2023
Sending data ...
Tue Oct 24 15:22:00 EAT 2023
Sending data ...
Tue Oct 24 15:22:10 EAT 2023
Sending data ...
Tue Oct 24 15:22:20 EAT 2023

The output is similar to our NIO example.