package kafka.network
import java.nio._
import java.nio.channels._
import kafka.utils._
import kafka.api.RequestOrResponse
@nonthreadsafe
private[kafka] class BoundedByteBufferSend(val buffer: ByteBuffer) extends Send {
private var sizeBuffer = ByteBuffer.allocate(4)
if(buffer.remaining > Int.MaxValue - sizeBuffer.limit)
throw new IllegalStateException("Attempt to create a bounded buffer of " + buffer.remaining + " bytes, but the maximum " +
"allowable size for a bounded buffer is " + (Int.MaxValue - sizeBuffer.limit) + ".")
sizeBuffer.putInt(buffer.limit)
sizeBuffer.rewind()
var complete: Boolean = false
def this(size: Int) = this(ByteBuffer.allocate(size))
def this(request: RequestOrResponse) = {
this(request.sizeInBytes + (if(request.requestId != None) 2 else 0))
request.requestId match {
case Some(requestId) =>
buffer.putShort(requestId)
case None =>
}
request.writeTo(buffer)
buffer.rewind()
}
def writeTo(channel: GatheringByteChannel): Int = {
expectIncomplete()
var written = channel.write(Array(sizeBuffer, buffer))
if(!buffer.hasRemaining)
complete = true
written.asInstanceOf[Int]
}
}