private static Observable < PingServiceHealth > mapToServiceHealth ( final String scope , final ServiceType type , final Observable < ? extends CouchbaseResponse > input , final AtomicReference < CouchbaseRequest > request , final long timeout , final TimeUnit timeUnit ) { return input . map ( new Func1 < CouchbaseResponse , PingServiceHealth > ( ) { @ Override public PingServiceHealth call ( CouchbaseResponse response ) { DiagnosticRequest request = ( DiagnosticRequest ) response . request ( ) ; String id = "0x" + Integer . toHexString ( request . localSocket ( ) . hashCode ( ) ) ; return new PingServiceHealth ( type , PingServiceHealth . PingState . OK , id , TimeUnit . NANOSECONDS . toMicros ( System . nanoTime ( ) - response . request ( ) . creationTime ( ) ) , request . localSocket ( ) , request . remoteSocket ( ) , scope ) ; } } ) . onErrorReturn ( new Func1 < Throwable , PingServiceHealth > ( ) { @ Override public PingServiceHealth call ( Throwable throwable ) { SocketAddress local = ( ( DiagnosticRequest ) request . get ( ) ) . localSocket ( ) ; SocketAddress remote = ( ( DiagnosticRequest ) request . get ( ) ) . remoteSocket ( ) ; String id = local == null ? "0x0000" : "0x" + Integer . toHexString ( local . hashCode ( ) ) ; if ( throwable instanceof TimeoutException ) { return new PingServiceHealth ( type , PingServiceHealth . PingState . TIMEOUT , id , timeUnit . toMicros ( timeout ) , local , remote , scope ) ; } else { LOGGER . warn ( "Error while running PingService for {}" , type , throwable ) ; return new PingServiceHealth ( type , PingServiceHealth . PingState . ERROR , id , TimeUnit . NANOSECONDS . toMicros ( System . nanoTime ( ) - request . get ( ) . creationTime ( ) ) , local , remote , scope ) ; } } } ) ; }
Helper method to perform the proper health conversion .