package kafka.api
import java.nio.ByteBuffer
import kafka.network.{BoundedByteBufferSend, RequestChannel}
import kafka.common.ErrorMapping
import org.apache.kafka.common.requests.{HeartbeatResponse, HeartbeatRequest}
import kafka.api.ApiUtils._
import kafka.network.RequestChannel.Response
import scala.Some
object HeartbeatRequestAndHeader {
def readFrom(buffer: ByteBuffer): HeartbeatRequestAndHeader = {
val versionId = buffer.getShort
val correlationId = buffer.getInt
val clientId = readShortString(buffer)
val body = HeartbeatRequest.parse(buffer)
new HeartbeatRequestAndHeader(versionId, correlationId, clientId, body)
}
}
case class HeartbeatRequestAndHeader(override val versionId: Short,
override val correlationId: Int,
override val clientId: String,
override val body: HeartbeatRequest)
extends GenericRequestAndHeader(versionId, correlationId, clientId, body, RequestKeys.nameForKey(RequestKeys.HeartbeatKey), Some(RequestKeys.HeartbeatKey)) {
override def handleError(e: Throwable, requestChannel: RequestChannel, request: RequestChannel.Request): Unit = {
val errorResponseBody = new HeartbeatResponse(ErrorMapping.codeFor(e.getClass.asInstanceOf[Class[Throwable]]))
val errorHeartBeatResponseAndHeader = new HeartbeatResponseAndHeader(correlationId, errorResponseBody)
requestChannel.sendResponse(new Response(request, new BoundedByteBufferSend(errorHeartBeatResponseAndHeader)))
}
}