THIS IS A TEST INSTANCE. ALL YOUR CHANGES WILL BE LOST!!!!
...
Code Block | ||||||
---|---|---|---|---|---|---|
| ||||||
/** The Observer interface. This class allows user to implement their own auditing solution. * Notice that the Observer may be used by multiple threads, so the implementation should be thread safe. */ public interface Observer extends Configurable { /* Record the request and response pair based on the given information. */ public void record(RequestAdapterRequestInfo request, ResponseAdapterResponseInfo response); /* Close the observer with timeout. * * @param timeout the maximum time to wait to close the observer. * @param unit The time unit. */ public void close(long timeout, TimeUnit unit); } |
...
Code Block | ||||||
---|---|---|---|---|---|---|
| ||||||
private def sendResponse(request: RequestChannel.Request,
responseOpt: Option[AbstractResponse],
onComplete: Option[Send => Unit]): Unit = {
// Update error metrics for each error code in the response including Errors.NONE
responseOpt.foreach(response => requestChannel.updateErrorMetrics(request.header.apiKey, response.errorCounts.asScala))
val response = responseOpt match {
case Some(response) =>
val responseSend = request.context.buildResponse(response)
// Convert RequestChannel.Request and AbstractResponse into their adapter instances which implements the request/response info interface
val requestAdapter = new RequestAdapter(request)
val responseAdapter = new ResponseAdapter(response)
// Invoke the record method on each observer on each request and response info instance. "observers" is a list of observers and each of
// them represents one user-provided implementation of the observer interface.
for (observer <- observers) {
try {
(observer: Observer).record(requestAdapter, responseAdapter)
} catch {
case e: Throwable => error(s"Observer ${(observer: Object).getClass} failed to record $request and $response", e)
}
}
val responseString =
if (RequestChannel.isRequestLoggingEnabled) Some(response.toString(request.context.apiVersion))
else None
new RequestChannel.SendResponse(request, responseSend, responseString, onComplete)
case None =>
new RequestChannel.NoOpResponse(request)
}
sendResponse(response)
} |
...