Global request interceptor

[sergeys-opera]

I couldn't find a way in Wire to intercept all the requests and handle their responses. Without that implementing things like authentication is much harder.

The way it works in io.grpc is that you implement and add a io.grpc.ClientInterceptor when creating a io.grpc.Channel. For example, the code authenticating requests might look like this (note that the existing auth token is invalidated on receiving Status.UNAUTHENTICATED from the server):

class GrpcAuthInterceptor (private val authTokenHolder: AuthTokenHolder) : ClientInterceptor {
    override fun <ReqT : Any, RespT : Any> interceptCall(
        method: MethodDescriptor<ReqT, RespT>,
        callOptions: CallOptions,
        next: Channel
    ): ClientCall<ReqT, RespT> = object : ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(
        next.newCall<ReqT, RespT>(method, callOptions)
    ) {
        override fun start(responseListener: Listener<RespT>, headers: Metadata) {
            val needsAuth = headers.needsAuth()
            if (!needsAuth) return super.start(responseListener, headers)

            val authToken = runBlocking { authTokenHolder.getAuthToken() }
            if (authToken != null) {
                headers.authenticate(authToken)
            }
            super.start(
                if (authToken != null) responseListener(authToken, responseListener) else responseListener,
                headers
            )
        }
    }

    private fun <RespT : Any> responseListener(
        authToken: String,
        original: ClientCall.Listener<RespT>
    ) = object : ForwardingClientCallListener.SimpleForwardingClientCallListener<RespT>(original) {
        override fun onClose(status: Status, trailers: Metadata) {
            super.onClose(status, trailers)
            if (status == Status.UNAUTHENTICATED) {
                runBlocking {
                    authTokenHolder.invalidate(authToken)
                }
            }
        }
    }
}

However, it's hard to do the same in Wire as there is no single point of request handling.

Here it was suggested to add an OkHttpClient interceptor and set the HTTP headers instead which IMO is not correct as it's a wrong application layer (HTTP vs gRPC). Moreover, it's hard to detect authentication errors as gRPC has its own status codes and always returns HTTP status code 200 (even if the request couldn't be authenticated according to gRPC, for example). This of course could be worked around by checking the grpc-status header but it doesn't look clean to me.

Am I missing something? Or how are you supposed to implement authentication (or any other generic request handling) in Wire?

[sergeys-opera]

Ended up with the following code:

suspend fun <S : Any, R : Any> GrpcCall<S, R>.executeAuthenticated(authTokenHolder: AuthTokenHolder, request: S): R {
    val token = authTokenHolder.getAuthToken() ?: return execute(request)

    requestMetadata += AUTH_HEADER to token
    return try {
        execute(request)
    } catch (e: GrpcException) {
        if (e.grpcStatus == GrpcStatus.UNAUTHENTICATED) {
            authTokenHolder.invalidate(token)
        }
        throw e
    }
}

executeAuthenticated method-extension has to be used instead of com.squareup.wire.GrpcCall#execute where authentication is needed.

There are two drawbacks with this solution:

1. It's hard to add more "interceptors" as the code invokes GrpcCall#execute.
2. The call sight needs to have access to a possibly internal AuthTokenHolder.

[sailquilt]

+1, would be great to have ClientInterceptor API for things like logging.

[oldergod]

That feels sad to do that if OkHttp already does it (very well)

[sergeys-opera]

That feels sad to do that if OkHttp already does it (very well)

However, it's done in a different level/layer which causes some inconvenience. For example, if you would want to log every request or response in OkHttp's interceptor the data would be decoded twice - once in the interceptor and once in the Wire's code. As mentioned above, you need to know gRPC protocol implementation details in order to handle the status codes while intercepting HTTP requests.

[sergeys-opera]

Would it be possible to expose some internal methods to facilitate response parsing in OkHttp interceptor? I'm primarily interested in GrpcResponse.grpcResponseToException

[sergeys-opera]

A more concrete issue with using OkHttp's interceptor is that the trailers might not be available at the point when an interceptor is invoked.

[oldergod]

Would opening GrpcResponse.grpcResponseToException help enough? I think that'd be fine to change its visibility to public

[sergeys-opera]

Would opening GrpcResponse.grpcResponseToException help enough?

Yes. It would be good to mention in the documentation that the response has to be consumed prior to calling GrpcResponse.grpcResponseToException (or let me know if it's a stupid idea and there is a better way to ensure that the response trailers are received).

This is how I handle gRPC status codes now:

    private fun Response.grpcStatus(): GrpcStatus? {
        val source = body?.source() ?: return null
        // Buffer the entire body. We need to read the whole body to ensure that the trailers are received.
        source.request(Long.MAX_VALUE)

        val grpcStatus = trailers()["grpc-status"] ?: header("grpc-status") ?: return null
        val grpcStatusInt = grpcStatus.toIntOrNull() ?: return null
        return GrpcStatus.get(grpcStatusInt)
    }