Implement Chat CLI Client

In the ChatClient.initChatStream(), call chatStreamService.chat(…): and assign it to this.toServer variable.

def initChatStream(jwtCallCredential:JwtCallCredential, clientOutput: String => Unit): Unit = {
  val streamObserver = new StreamObserver[ChatMessageFromServer] {
   override def onError(t: Throwable): Unit = {
     logger.error("gRPC error", t)
     shutdown()
   }

   override def onCompleted(): Unit = {
     logger.error("server closed connection, shutting down...")
     shutdown()
   }

   override def onNext(chatMessageFromServer: ChatMessageFromServer): Unit = {
     try {
       clientOutput(s"${chatMessageFromServer.getTimestamp.seconds} ${chatMessageFromServer.from}> ${chatMessageFromServer.message}")
     }
     catch {
       case exc: IOException =>
         logger.error("Error printing to console", exc)
       case exc: Throwable => logger.error("grpc exception", exc)
     }
   }

   optChatChannel.foreach { chatChannel =>
    val chatStreamService = ChatStreamServiceGrpc.stub(chatChannel).withCallCredentials(jwtCallCredential)
    val toServer = chatStreamService.chat(streamObserver)
    optToServer = Some(toServer)
   }
 }

When the chat messages arrive from the server, onNext will be called. Print out the message.

override def onNext(chatMessageFromServer: ChatMessageFromServer): Unit = {
  try {
    clientOutput(s"${chatMessageFromServer.getTimestamp.seconds} ${chatMessageFromServer.from}> ${chatMessageFromServer.message}")
  }
  catch {
    case exc: IOException =>
      logger.error("Error printing to console", exc)
    case exc: Throwable => logger.error("grpc exception", exc)
  }
}

When the server throws an error or close the connection, shutdown the client:

override def onError(t: Throwable): Unit = {
  logger.error("gRPC error", t)
  shutdown()
}

override def onCompleted(): Unit = {
  logger.error("server closed connection, shutting down...")
  shutdown()
}

Finally, implement ChatClient.sendMessage(…) method. Every time a user presses enter, it’ll call this method to send the message out to the server:

  1. Check that toServer observer is not null
  2. Then, call toServer.onNext(…) to send the message to the server to be broadcasted to the room
def sendMessage(room: String, message: String): Unit = {
  logger.info("sending chat message")
  // call toServer.onNext(...)
  optToServer match {
    case Some(toServer) =>
      val chatMessage = ChatMessage(MessageType.TEXT, room, message)
      toServer.onNext(chatMessage)
    case None =>
      logger.info("Not Connected")
  }
}

Run the Chat Client

Run the chat client in multiple terminal windows.

In one terminal window:

$ sbt chatclient/run
...
STATE: CurrentState(STARTED,,,)
/login [username] | /quit
-> /login admin
[run-main-0] INFO ChatClient - processing login user
password> admin
...
admin-> /join beta

In another terminal window:

$ sbt chatclient/run
...
STATE: CurrentState(STARTED,,,)
/login [username] | /quit
-> /login hydro
[run-main-0] INFO ChatClient - processing login user
password> hydro
...
hydro-> /join beta
...
hydro

Back in the first terminal window you should see:

hydro-> /join beta