Akka/Scala example source code file (WithUdpSend.scala)

This example Akka source code file (WithUdpSend.scala) is included in my "Source Code Warehouse" project. The intent of this project is to help you more easily find Akka and Scala source code examples by using tags.

All credit for the original source code belongs to akka.io; I'm just trying to make examples easier to find. (For my Scala work, see my Scala examples and tutorials.)

Akka tags/keywords

actor, akka, channelregistration, channelwritable, commandfailed, datagramchannel, io, receive, send, tracelogging, udpext, withudpsend

The WithUdpSend.scala Akka example source code

/**
 * Copyright (C) 2009-2014 Typesafe Inc. <http://www.typesafe.com>
 */
package akka.io

import java.nio.channels.{ SelectionKey, DatagramChannel }
import akka.actor.{ ActorRef, ActorLogging, Actor }
import akka.io.Udp.{ CommandFailed, Send }
import akka.io.SelectionHandler._

/**
 * INTERNAL API
 */
private[io] trait WithUdpSend {
  me: Actor with ActorLogging ⇒

  private var pendingSend: Send = null
  private var pendingCommander: ActorRef = null
  // If send fails first, we allow a second go after selected writable, but no more. This flag signals that
  // pending send was already tried once.
  private var retriedSend = false
  private def hasWritePending = pendingSend ne null

  def channel: DatagramChannel
  def udp: UdpExt
  val settings = udp.settings

  import settings._

  def sendHandlers(registration: ChannelRegistration): Receive = {
    case send: Send if hasWritePending ⇒
      if (TraceLogging) log.debug("Dropping write because queue is full")
      sender() ! CommandFailed(send)

    case send: Send if send.payload.isEmpty ⇒
      if (send.wantsAck)
        sender() ! send.ack

    case send: Send ⇒
      pendingSend = send
      pendingCommander = sender()
      doSend(registration)

    case ChannelWritable ⇒ if (hasWritePending) doSend(registration)
  }

  private def doSend(registration: ChannelRegistration): Unit = {
    val buffer = udp.bufferPool.acquire()
    try {
      buffer.clear()
      pendingSend.payload.copyToBuffer(buffer)
      buffer.flip()
      val writtenBytes = channel.send(buffer, pendingSend.target)
      if (TraceLogging) log.debug("Wrote [{}] bytes to channel", writtenBytes)

      // Datagram channel either sends the whole message, or nothing
      if (writtenBytes == 0) {
        if (retriedSend) {
          pendingCommander ! CommandFailed(pendingSend)
          retriedSend = false
          pendingSend = null
          pendingCommander = null
        } else {
          registration.enableInterest(SelectionKey.OP_WRITE)
          retriedSend = true
        }
      } else {
        if (pendingSend.wantsAck) pendingCommander ! pendingSend.ack
        retriedSend = false
        pendingSend = null
        pendingCommander = null
      }
    } finally {
      udp.bufferPool.release(buffer)
    }
  }
}