connectrpc / connect-kotlin Goto Github PK

Hey, I am using the Connect library in my app which works pretty good so far. We have to stick to v0.5.0 for now as some other dependency need protobuf 3.25 which makes it incompatible with 0.6.0 for now.

However, I get some errors from users via Sentry like this:

java.lang.IllegalStateException: trailers not available
    at okhttp3.Response.trailers(Response.kt:180)
    at com.connectrpc.okhttp.ConnectOkHttpClient$unary$1.onResponse(ConnectOkHttpClient.kt:160)
    at okhttp3.internal.connection.RealCall$AsyncCall.run(RealCall.kt:519)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:644)
    at java.lang.Thread.run(Thread.java:1012)

This seems to be the case whenever the network connection is lost / flaky.

I am executing the call to the network function inside a normal try catch block. The stacktrace also suggests that I cannot wrap it in a way that catches the crash as it directly originates from the OkHttp threadpool rather than my code.

Can you have a look at it please? And maybe you have an idea for a quick workaround as this occurs quite regularly.

The streaming interfaces' send() method should return an object/enum to signal to the user if the underlying stream is closed (SendResult potentially).

Additionally, some nice ergonomic methods for the stream would be to expose an isClosed() method for the user to check if the underlying stream is available.

The java_multiple_files option is described like this in the protobuf documentation:

... For each .proto file input, the compiler creates a wrapper .java file containing a Java class which represents the .proto file itself.

If the .proto file contains a line like the following:

option java_multiple_files = true;

Then the compiler will also create separate .java files for each of the classes/enums which it will generate for each top-level message, enumeration, and service declared in the .proto file.

Otherwise (i.e. when the java_multiple_files option is disabled; which is the default), the aforementioned wrapper class will also be used as an outer class, and the generated classes/enums for each top-level message, enumeration, and service declared in the .proto file will all be nested within the outer wrapper class. Thus the compiler will only generate a single .java file for the entire .proto file.

The grpc-java and grpc-kotlin libraries both honor this option. For the given .proto file, the following outputs are produced by grpc-java:

syntax = "proto3";

package com.example;

option java_package = "com.example";

service EventService {
    rpc Start (StartRequest) returns (StartResponse);
}

message StartRequest { }

message StartResponse { }

With java_multiple_files = false

public suspend fun start(request: com.example.EventsServiceProto.StartRequest, headers: Metadata = Metadata()): com.example.EventsServiceProto.StartResponse = unaryRpc(
  channel,
  EventServiceGrpc.getStartMethod(),
  request,
  callOptions,
  headers
)

With java_multiple_files = true

public suspend fun start(request: com.example.StartRequest, headers: Metadata = Metadata()): com.example.StartResponse = unaryRpc(
  channel,
  EventServiceGrpc.getStartMethod(),
  request,
  callOptions,
  headers
)

However, protoc-gen-connect-kotlin generates identical output irrespective of the value of java_multiple_files:

public override suspend fun start(request: StartRequest, headers: Headers): ResponseMessage<StartResponse> = client.unary(
  request,
  headers,
  MethodSpec(
  "live.v1.EventService/Start",
    com.example.StartRequest::class,
    com.example.StartResponse::class,
  ),
)

This means that the Request and Response classes are not found when trying to compile the application.

The connect-kotlin generator probably should check this option and generate compatible code.

When testing code that has a dependency on one of the API clients/endpoints that connect-kotlin generates, consumers have to rely on mocking frameworks to stub out the generated calls and replace their responses.

This generally works, due to the maturity of mocking frameworks on the JVM, but as an alternative (for projects that don't want to import/use mocking frameworks), buf/the BSR could generate fake API client implementations for testing purposes (similar to how the connect-swift-mocks plugin works).

The HTTP/2 definition of a "trailers only" response (per gRPC protocol spec) has no data frames: the response is done with the single response header frame.

This must be relaxed a little for gRPC-Web, which does not require HTTP/2. In existing implementations, a "trailers only" response is one that has no body and no trailers (the headers are interpreted as trailers).

This client, on the other hand, eagerly searches for a "grpc-status" header in the response headers, and assumes it is a "trailers only" response if it is present -- even if the response includes a body or other trailers.

To better align with other implementations and to more closely align to the definition of a "trailers only" response in the gRPC spec, this client should only consider a response to be a "trailers only" response when there is no body and no trailers. If it receives a response that includes "grpc-status" in the headers, but also has a body (or any trailers), that initial status header should be ignored.

The current behavior does not typically issues in practice, but could cause incorrect interpretation of responses in misbehaving servers or in servers that inadvertently allow a (possibly malicious) user-provided payload to set extra response headers.

When the network inspector in Android Studio is active while I perform a connect streaming call, the app just crashes with the following error:

E  FATAL EXCEPTION: OkHttp Dispatcher
	Process: com.connectrpc.examples.android, PID: 11743
	java.lang.IllegalStateException: sink already folded
		at okio.Pipe.fold(Pipe.kt:177)
		at com.connectrpc.okhttp.PipeDuplexRequestBody.writeTo(OkHttpStream.kt:211)
		at okhttp3.internal.http.CallServerInterceptor.intercept(CallServerInterceptor.kt:58)
		at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.kt:109)
		at com.android.tools.profiler.agent.okhttp.OkHttp3Interceptor.intercept(OkHttp3Interceptor.java:57)
		at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.kt:109)
		at com.android.tools.appinspection.network.okhttp.OkHttp3Interceptor.intercept(OkHttp3Interceptor.kt:52)
		at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.kt:109)
		at com.android.tools.profiler.agent.okhttp.OkHttp3Interceptor.intercept(OkHttp3Interceptor.java:57)
		at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.kt:109)
		at com.android.tools.appinspection.network.okhttp.OkHttp3Interceptor.intercept(OkHttp3Interceptor.kt:52)
		at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.kt:109)
		at okhttp3.internal.connection.ConnectInterceptor.intercept(ConnectInterceptor.kt:34)
		at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.kt:109)
		at okhttp3.internal.cache.CacheInterceptor.intercept(CacheInterceptor.kt:95)
		at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.kt:109)
		at okhttp3.internal.http.BridgeInterceptor.intercept(BridgeInterceptor.kt:83)
		at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.kt:109)
		at okhttp3.internal.http.RetryAndFollowUpInterceptor.intercept(RetryAndFollowUpInterceptor.kt:76)
		at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.kt:109)
		at okhttp3.internal.connection.RealCall.getResponseWithInterceptorChain$okhttp(RealCall.kt:201)
		at okhttp3.internal.connection.RealCall$AsyncCall.run(RealCall.kt:517)
		at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1137)
		at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:637)
		at java.lang.Thread.run(Thread.java:1012)

To reproduce, you can simply run the Eliza example from this repo, open the network inspector (wait until it is fully loaded) and open the Connect - bidirectional streaming example. The app crashes immediately with the error above.

Are there any plans to make Connect a KMP friendly library? Personally interested in JVM, Android and iOS targets.

This change in #22 prevents the stream from staying open for long-lived connections.
For example, when handling user chat input, version 0.1.2 keeps the connection open while 0.1.3 and newer close the connection after a message is sent because the buffer is "temporarily" out of data until the user types the next message.

    fun forConsume(buffer: Buffer) {
        try {
            if (bufferedSink.isOpen) {
                bufferedSink.writeAll(buffer)
                bufferedSink.flush()
            }
-        } catch (e: Throwable) {
+        } finally {
            close()
        }
    }

I think that what should actually be here is something like this:

    fun forConsume(buffer: Buffer) {
        try {
            if (bufferedSink.isOpen) {
                bufferedSink.writeAll(buffer)
                bufferedSink.flush()
            }
+       } catch (e: Throwable) {
            close()
+           throw e;
        }
    }

In #130, we're simplifying the API for streaming calls to make it much simpler for consumers to receive messages and not build custom error handling. Additionally, the previous API made it seem like there may be multiple headers/trailers per response, instead of one for the entire stream. However, it is no longer possible via the result channel to read headers/trailers. They are still available via interceptors and ConnectException (for errors) however.

There are a couple of options for how to expose headers/trailers to streaming consumers:

1. Expose responseHeaders: Future<Headers> and responseTrailers: Future<Trailers> on BidirectionStreamInterface/ClientOnlyStreamInterface/ServerOnlyStreamInterface.
2. Expose the same as (1) but use https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-deferred/ instead of Future.

Additionally, https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/ has been mentioned as a possibility.

Reference:

• connect-go: https://github.com/connectrpc/connect-go/blob/96effedc8ac84da9f49392e5f7bdfee2e648247b/client_stream.go#L251-L267

There have been issues in the past w/ crosstests, Maven, and other build flows where the Makefile is used but is missing required code generation dependencies (you have to manually add the generate prerequisite - see #95 for an example).

Ideally code generation would work in connect-kotlin so any user could run ./gradlew build and get a successful build without having to find the appropriate make targets. It could configure this so the compileKotlin gradle task requires code generation first.

This should be pretty easy to automate (we might even want to look at the buf gradle plugin for this) and it would reduce maintenance long term.

bufbuild/protovalidate-java#46 is an example of this migration for another project.

Instead of downloading protoc, we should be able to migrate the code generation used in this project to work with remote plugins. This might be work that could be picked up alongside #228.

Hello everyone,

First of all, thank you for open-sourcing this awesome piece of software. We're currently using the whole Connect stack for a project at work and everything has gone smoothly so far.

That is, until we started using client-only streams in-between our Android app and our back-end (Go). This simply didn't work in all versions up to and including 0.4.0.

The good news is, we've came across @jhump's recent contribution (868181b), tested it and it fixed everything 🎉

When can we expect a new version of connect-kotlin including these changes?

Best regards

I'm trying to establish a stream connection to a remote and deal with connection issues.
My code to establish the response channel currently looks like this:

val receiveData = sequencerClient.receiveData() // it's a ServerOnlyStreamInterface
val receiveDataRequest = receiveDataWithPreloadSelector(preloadSelector) // just the request input
receiveData.sendAndClose(receiveDataRequest) // always succeeds, even if the remote is not available
receiveData.responseChannel()

if the remote service is not available, I would have expected the invocation of sendAndClose to raise an error. However, it looks like sendAndClose doesn't actually send anything. Only when invoking receiveCatching on the response channel do I get an exception that the remote is not available.

However, I really want to have different error handling behaviour for establishing the initial connection and dealing with subsequent issues: if it cannot establish the connection, it's not so bad - but if something breaks while streaming data of an existing connection, it needs to handle it more urgently.

Of course I can introduce a new state variable that indicates whether something was already read etc. But I find it quite irritating that a method called sendAndClose doesn't raise an exception when it cannot send anything.

I opened #12 because there were problems when the base URL ended with a /. Unfortunately, the URL handling changes in #17 now cause the application to throw an exception when the URL DOES NOT end with a trailing slash.

I discovered this when working with a base URL that ends with a port number https://example.com:8443 and endpoint live.v1.EventService/Status, which crashes with this exception:

java.lang.NumberFormatException: For input string: "443live.v1.EventService"
	at java.lang.Integer.parseInt(Integer.java:747)
	at java.lang.Integer.parseInt(Integer.java:865)
	at java.net.URLStreamHandler.parseURL(URLStreamHandler.java:246)
	at java.net.URL.<init>(URL.java:630)
	at java.net.URL.<init>(URL.java:498) 
	at java.net.URL.<init>(URL.java:447) 
	at java.net.URI.toURL(URI.java:1085) 
	at build.buf.connect.impl.ProtocolClient.urlFromMethodSpec(ProtocolClient.kt:206) 
	at build.buf.connect.impl.ProtocolClient.unary(ProtocolClient.kt:56) 
	at build.buf.connect.impl.ProtocolClient.unary(ProtocolClient.kt:104) 
	at live.saferworld.api.live.v1.EventServiceClient.status(EventServiceClient.kt:52) 
	at live.saferworld.api.live.v1.EventServiceClientInterface$DefaultImpls.status$default(EventServiceClientInterface.kt:20) 
// etc.

It looks like this comes from ProtocolClient.urlFromMethodSpec using URI.resolve, based on these results from the Kotlin playground:

println(URI("https://example.com").resolve("live.v1.EventService/Status"));
// WRONG: https://example.comlive.v1.EventService/Status
println(URI("https://example.com/").resolve("live.v1.EventService/Status"));
// RIGHT: https://example.com/live.v1.EventService/Status
println(URI("https://example.com:8443").resolve("live.v1.EventService/Status"));
// WRONG: https://example.com:443live.v1.EventService/Status
println(URI("https://example.com:8443/").resolve("live.v1.EventService/Status"));
// RIGHT: https://example.com:443/live.v1.EventService/Status

In #12 I had suggested using the URL(URL context, String spec) constructor (e.g. baseUrl = URL(host) in the ProtocolClientConfig and val url = URL(config.baseUrl, methodSpec.path)), but I see that can be an issue if a path is provided without a trailing slash, like this:

println(URI("https://example.com/test").resolve("live.v1.EventService/Status"));
// WRONG: https://example.com/live.v1.EventService/Status
println(URI("https://example.com/test/").resolve("live.v1.EventService/Status"));
// RIGHT: https://example.com/test/live.v1.EventService/Status

I suspect the simplest solution would be this, in ProtocolClientConfig.init:

baseUri = URI(if (host.endsWith('/')) host else "${host}/")

As far as I can tell, there is no situation in which this would cause problems.

I am not sure how this was able to get past the tests that were added in #17, because I see there are tests that use both with and without a slash. I wonder if there should be tests that verify that the URL is actually what the URL is expected to be.

I cant receive StreamResult.Message, only StreamResult.Headers and StreamResult.Complete are available.

        Log.d("send", reply)
        val stream = client.sendMessageStream()

        stream.send(createRequest())
        Log.d("send", "finish send")
        for (response in stream.resultChannel()) {
            Log.d("msg", "new part message ${response}")

            if (response.connectError() != null) {
                Log.e("error", "connect error")
            }
            if (response.error != null) {
                Log.e("error", response.error.toString())
            }
            when (response) {
                is StreamResult.Headers -> {
                    Log.d("msg", "header ${response.headers}")
                }

                is StreamResult.Message -> {
                    Log.d("msg", response.message.reply)
                }

                is StreamResult.Complete -> {
                    Log.d("msg", "complete ${response.code}")
                }
            }
        }

ProtocolClientConfig is a good example; Java consumers currently have to specify each parameter's default value manually.

When gzip compression is enabled and compressed, empty responses are returned, ConnectInterceptors response parsing fails due to GzipCompressionPool.decompress not being able to handle empty Buffers.

It's unclear to me if this is actually a valid response message state that connect-kotlin should handle, or if this is just a symptom of our QA environment being in a weird state serving back invalid responses.

Feel free to close if this is invalid

Description

I am currently using connect-kotlin with streaming calls in combination with the debugging tool Flipper and their network plugin. The problem is that Flipper manipulates these streaming calls in a way that makes them contain a body of 0 bytes which the server can obviously not process. Here is an example of what that looks like:

As you can see, the first call fails because it has a body size of 0B. However, the second call which is exactly the same, succeeds. So I would assume that it is the result of a race condition.

I think that the code responsible for this should be here. When timing is bad, connect just wants to read the request body when Flipper is performing the copy operation which has the side effect that the body is empty for a very short period of time. I would guess that the pipe you use is not directly compatible with that approach.

In the best case, this race condition is solved as when it works, Flipper is super useful to debug streaming calls.

To Reproduce

To reproduce, I created a fork of the connect-kotlin repository and added flipper to it here. To run, clone the repo, cd into the directory and run $ make generate. After that you can open Android Studio and execute examples.android. When the app is opened, select Connect - bidirectional streaming from the menu. In Flipper you will see that the request body is 0B:

Environment

Flipper 0.236.0
connect-kotlin 0.3.1

Hi,

One of the most annoying issues with GRPC in Kotlin is that the generated kotlin code does not work with optional.

It generates the following code:

    /**
     * `optional uint32 min_transaction_amount_in_cents = 5 [json_name = "minTransactionAmountInCents"];`
     */
    public var minTransactionAmountInCents: kotlin.Int
      @JvmName("getMinTransactionAmountInCents")
      get() = _builder.getMinTransactionAmountInCents()
      @JvmName("setMinTransactionAmountInCents")
      set(value) {
        _builder.setMinTransactionAmountInCents(value)
      }
    /**
     * `optional uint32 min_transaction_amount_in_cents = 5 [json_name = "minTransactionAmountInCents"];`
     */
    public fun clearMinTransactionAmountInCents() {
      _builder.clearMinTransactionAmountInCents()
    }
    /**
     * `optional uint32 min_transaction_amount_in_cents = 5 [json_name = "minTransactionAmountInCents"];`
     * @return Whether the minTransactionAmountInCents field is set.
     */
    public fun hasMinTransactionAmountInCents(): kotlin.Boolean {
      return _builder.hasMinTransactionAmountInCents()
    }

Which means that you need to think about using the has method to know if your field is present or not, because they set the default value. It's not the case with swift, rust or javascript
When I saw that connect-kotlin was out, I was hopeful that you would have fixed this problem, but I saw that connectrpc/kotlin depended on protocolbuffers/kotlin and figured you would still have this issue. Confirmed after trying it.

Do you plan on fixing it or is it a behaviour you're okay with? Do you know if there's a solution (besides rewriting the files)?

I noticed that my Android app was using URLs with a double slash after the hostname, like this:

https://server.example.com//com.example.v1.ExampleService/ExampleEndpoint

I realized that this is happening because I store my server base URL as an okhttp3.HttpUrl object, and then create the ProtocolClientConfig like this:

fun createConnectClient(): ProtocolClientInterface {
    return ProtocolClient(
        httpClient = ConnectOkHttpClient(),
        ProtocolClientConfig(
            host = server.toString(),
            serializationStrategy = GoogleJavaLiteProtobufStrategy(),
            networkProtocol = NetworkProtocol.CONNECT,
            interceptors = listOf(
                { AuthenticationInterceptor(User.bearerToken) },
                { LoggingInterceptor() },
            )
        ),
    )
}

For now, I work around this issue by manually building the host URL string, like this:

fun createConnectClient(): ProtocolClientInterface {
    return ProtocolClient(
        httpClient = ConnectOkHttpClient(),
        ProtocolClientConfig(
            // We cannot use `server.toString()` because it adds a trailing slash
            host = "${server.scheme}://${server.host}:${server.port}",
            serializationStrategy = GoogleJavaLiteProtobufStrategy(),
            networkProtocol = NetworkProtocol.CONNECT,
            interceptors = listOf(
                { AuthenticationInterceptor(User.bearerToken) },
                { LoggingInterceptor() },
            )
        ),
    )
}

However, I do not think this is a good long-term solution, for three reasons:

1. It is confusing for future developers who read the code and need to understand why we did this (requiring extra comments to explain).
2. In general, it would be nice to have some kind of validation that a proper URL has been provided.
3. The way that ProtocolClient builds a URL by concatenating strings (url = URL("${config.host}/${methodSpec.path}")) seems fragile.

I suggest that the host be stored as a URL object instead of as a String. Doing that fixes all three of my issues listed above, as follows:

1. Constructing the URL object from a String doesn't care if there's a trailing slash or not (or, in my case, using the toUrl() method on the HttpUrl object).
2. If an invalid URL is provided, the exception will be thrown immediately on construction instead of waiting until execution to fail.
3. You can build new URLs without string concatenation using the two-argument URL constructor that takes a base URL as "context": url = URL(config.host, methodSpec.path)

Regarding the last point about building the URL with a context, note that all five of the following invocations produce identical output:

URL("https://example.com/test")
URL(URL("https://example.com/"), "/test"))
URL(URL("https://example.com/"), "test"))
URL(URL("https://example.com"), "/test"))
URL(URL("https://example.com"), "test"))

Currently, the client assumes the response uses the correct content-type if it has a "200 OK" status code. To prevent strange issues in the face of misbehaving servers or middle-boxes, the client should actually verify the content-type.

If the content-type does not appear to be a valid RPC response (i.e. incorrect prefix), it should consider it an unknown error. If the content-type does appear to be a valid RPC response (correct prefix) but indicates the wrong codec (i.e. different than the request codec), it should consider it an internal error (since this indicates an internal problem in the server implementation).

Currently, this library relies wholly on the underlying HTTPClientInterface implementation handling timeouts. That poses two problems:

1. The only implementation of HTTPClientInterface provided is based on okhttp, which does not enforce timeouts on full-duplex bidirectional operations after the stream is established.
2. It should not be up to this layer to handle deadline propagation -- communicating the timeout to the server, so it can attenuate the deadline as it processes the request. That is a protocol concern (and Connect and gRPC/gRPC-Web both handle it slightly differently). Currently, timeout headers are never sent, so this client does not attempt to propagate deadlines to the server.

So to address these deficiencies:

1. There needs to be a mechanism for configuring timeouts other than the HTTPClientInterface implementation. This could be in the form of default timeouts in ProtocolClientConfig; it could be some way to specify timeouts on a per-RPC basis; or it could be both.
2. When such a timeout is configured, the protocol handling code needs to encode the timeout in a request header to communicate the deadline to the server.

The underlying HTTPClientInterface might still impose its own timeouts. So it's an open question as to whether this needs to be considered in the timeout handling. For example, if the user sets a 20 second timeout for an RPC, but the underlying HTTP client has a 10 second timeout configured, should the library take this into consideration and communicate a 10 second deadline to the server? If so, then we may need to add methods to HTTPClientInterface (or create sub-interfaces that can be optionally implemented) so that the framework can ask the client about its timeouts, in order to correctly consider them.

Alternatively, instead of the framework asking the client about its timeout, it could tell the client what timeout to use (similarly to how #14 proposes that the framework may need to tell the client what protocols to support, depending on the RPC protocol and URL scheme used for a particular operation).

I was wondering why the GoogleJava(Lite)ProtobufStrategy is available for both the java and the javalite runtime, but the GoogleJavaJSONStrategy can only be used with the java runtime. Is there a technical reason, why the json equivalent cannot be added to the extensions?

The gRPC libraries for Java/Kotlin (and most other languages) provide a usePlaintext() builder method that tells the library to communicate using HTTP/2 over Cleartext (h2c). This is mostly useful for communicating with development servers running on the developers' PC and for certain types of automated testing.

Because connect-kotlin uses a full URL instead of separate host and port arguments, we can check whether the scheme is http or https to determine whether to enable plaintext support.

We for want to keep HTTP/1.1 support for unary connect calls, which is similar to the discussion in #13 about separate client options for unary vs. streaming. However, gRPC requires HTTP/2 for unary calls too, so we probably need to check the chosen protocol before setting this option.

Here's what I'm using for this right now (built on top of my code sample in #13). Note that this cannot use automatic detection based on the URL scheme and network protocol because the scheme is not available when the client is created. Instead, the constructor has two additional arguments:

class PingingConnectClient(client: OkHttpClient, usePlaintext: Boolean, networkProtocol: NetworkProtocol): HTTPClientInterface {
    private val internalUnaryClient = ConnectOkHttpClient(client.newBuilder()
        .apply {
            // Unary gRPC uses HTTP/2, but connect can still use HTTP/1.1
            if (networkProtocol == NetworkProtocol.GRPC && usePlaintext) {
                protocols(listOf(Protocol.H2_PRIOR_KNOWLEDGE))
            }
        }
        .build())

    private val internalStreamClient = ConnectOkHttpClient(client.newBuilder()
        .pingInterval(30, TimeUnit.SECONDS)
        .readTimeout(0, TimeUnit.SECONDS)
        .apply {
            // Streaming always must use HTTP/2
            if (usePlaintext) {
                protocols(listOf(Protocol.H2_PRIOR_KNOWLEDGE))
            }
        }
        .build())

    override fun unary(request: HTTPRequest, onResult: (HTTPResponse) -> Unit): Cancelable {
        return internalUnaryClient.unary(request, onResult)
    }

    override fun stream(
        request: HTTPRequest,
        onResult: suspend (StreamResult<Buffer>) -> Unit
    ): Stream {
        return internalStreamClient.stream(request, onResult)
    }
}

I suspect that it makes the most sense to implement this in ConnectOkHttpClient as follows (not tested), but I'm not totally sure:

class ConnectOkHttpClient(
    val client: OkHttpClient = OkHttpClient()
) : HTTPClientInterface {
    override fun unary(request: HTTPRequest, onResult: (HTTPResponse) -> Unit): Cancelable {
        val unaryClient = client.newBuilder().apply {
            if (request.url.protocol == "http" && request.contentType.startsWith("application/grpc")) {
                protocols(listOf(Protocol.H2_PRIOR_KNOWLEDGE))
            }
        }.build()
        // The rest of the existing function here, but replace `client.newCall` with `unaryClient.newCall`
    }

    override fun stream(request: HTTPRequest, onResult: suspend (StreamResult<Buffer>) -> Unit): Stream {
        val streamClient = client.newBuilder().apply {
            if (request.url.protocol == "http") {
                protocols(listOf(Protocol.H2_PRIOR_KNOWLEDGE))
            }
        }.build()
        return client.initializeStream(request, onResult)
    }

Background:

I see that this project is using https://github.com/vanniktech/gradle-maven-publish-plugin. This should publish sources.

I see that com.connectrpc:connect-kotlin-okhttp:0.6.1 is published on mavenCentral() here https://central.sonatype.com/artifact/com.connectrpc/connect-kotlin-okhttp.

I have checked https://search.maven.org/artifact/com.connectrpc/connect-kotlin-okhttp/0.6.1/jar and https://repo1.maven.org/maven2/com/connectrpc/connect-kotlin-okhttp/0.6.1/ and sources are published.

See this link https://repo1.maven.org/maven2/com/connectrpc/connect-kotlin-okhttp/0.6.1/connect-kotlin-okhttp-0.6.1-sources.jar.

Error / Repro:

When I try to use the IDE to manually download the sources I see the following error:

Task 'ijDownloadSources18f23d87-b7a' not found in root project 'android-gif-search' and its subprojects.

* Try:
> Run gradle tasks to get a list of available tasks.
> For more on name expansion, please refer to https://docs.gradle.org/8.7/userguide/command_line_interface.html#sec:name_abbreviation in the Gradle documentation.
> Run with --stacktrace option to get the stack trace.
> Run with --info or --debug option to get more log output.
> Get more help at https://help.gradle.org.
BUILD FAILED in 163ms

or image:

I see the same when stepping into the code for debugging and I click on "Download Source":

Going forward:

Is this an IDE problem or the way the sources are being published?

Currently, the plugin is only published to the BSR. It would be nice for users of Connect-Kotlin to also be able to consume the plugin independently from the BSR. The options for this are:

1. Publish directly to maven as another artifact: build.buf:connect-kotlin:protoc-gen-connect-kotlin
2. Add an additional artifact within the releases which is the plugin jar

Result return on stream sends: #21 allows us to propagate errors from send to the user.

The main error that needs to be surfaced is for main thread sends on Android applications.

I've noticed a potential issue with the urlFromMethodSpec method in the com.connectrpc.impl.ProtocolClient class. It seems to have a problem when concatenating the domain and path.

For example:
If baseUri is "https://hello.com/api" and methodSpec.path is "world",

the expected result should be:
https://hello.com/api/world,
but the actual result is:
https://hello.com/world.

The "api" part is omitted. I'm wondering if this is a feature or a bug?

The default okhttp readTimeout is 10 seconds, which is too short for many uses of streaming RPCs. As an alternative way to detect broken connections, HTTP/2 provides a PING mechanism. okhttp has pings disabled by default, but they can be enabled by calling pingInterval() on the builder. Streaming calls can set readTimeout to 0 to disable the read timeout and rely only on the ping timeout, if desired.

The problem is that there is not an easy way to change the okhttp settings for unary vs. streaming calls, and we still want to have a read timeout for the unary calls.

I thought it would be easiest to fix this by extending ConnectOkHttpClient to customize the client object and then call the parent method using the customized client, but ConnectOkHttpClient would need to be open instead of final to allow this.

I ended up doing this, which feels like a nasty hack:

class PingingConnectClient(client: OkHttpClient): HTTPClientInterface {
    private val internalUnaryClient = ConnectOkHttpClient(client)
    private val internalStreamClient = ConnectOkHttpClient(client.newBuilder()
        .pingInterval(30, TimeUnit.SECONDS)
        .readTimeout(0, TimeUnit.SECONDS)
        .build())

    override fun unary(request: HTTPRequest, onResult: (HTTPResponse) -> Unit): Cancelable {
        return internalUnaryClient.unary(request, onResult)
    }

    override fun stream(
        request: HTTPRequest,
        onResult: suspend (StreamResult<Buffer>) -> Unit
    ): Stream {
        return internalStreamClient.stream(request, onResult)
    }
}

Generating a .proto file with no package defined causes an error on generation:

Failure: plugin "buf.build/bufbuild/connect-kotlin:v0.1.1" exited with non-zero status 1: Exception in thread "main" java.lang.IllegalArgumentException: couldn't make a guess for .HelloRequest
        at com.squareup.kotlinpoet.ClassName$Companion.bestGuess(ClassName.kt:205)
        at build.buf.protocgen.connect.Generator.interfaceMethods(Generator.kt:128)
        at build.buf.protocgen.connect.Generator.serviceClientInterface(Generator.kt:112)
        at build.buf.protocgen.connect.Generator.parseFile(Generator.kt:89)
        at build.buf.protocgen.connect.Generator.generate(Generator.kt:73)
        at build.buf.protocgen.connect.internal.Plugin.run(Plugin.kt:121)
        at build.buf.protocgen.connect.internal.Plugin.run$default(Plugin.kt:86)
        at build.buf.protocgen.connect.Main$Companion.main(Main.kt:23)
        at build.buf.protocgen.connect.Main.main(Main.kt)

Currently, the javadoc artifacts published to Maven central don't have any content. We should follow the steps referenced here to publish real javadocs: https://kotlinlang.org/docs/dokka-gradle.html#build-javadoc-jar

My gradle config

android {
    namespace = "xxx"
    compileSdk = 33

    signingConfigs {
        ...
    }

    defaultConfig {
        applicationId = "xxx"
        minSdk = 26
        targetSdk = 33
        versionCode = 1
        versionName = "1.0.0-debug-v10"

        testInstrumentationRunner = "androidx.test.runner.AndroidJUnitRunner"
        vectorDrawables {
            useSupportLibrary = true
        }
        resourceConfigurations.addAll(listOf("en", "ja", "zh-rCN", "zh-rTW"))
    }

    val useMinifyInDebug = hasProperty("minifyInDebug")

    buildTypes {
        debug {
            isMinifyEnabled = useMinifyInDebug
            isShrinkResources = useMinifyInDebug
            isDebuggable = true
            configure<CrashlyticsExtension> {
                mappingFileUploadEnabled = useMinifyInDebug
            }
            signingConfig = signingConfigs.getByName("debug")
            proguardFiles(
                getDefaultProguardFile("proguard-android-optimize.txt"),
                "proguard-rules.pro"
            )
        }
        release {
            isMinifyEnabled = true
            isShrinkResources = true
            configure<CrashlyticsExtension> {
                mappingFileUploadEnabled = true
            }
            signingConfig = signingConfigs.getByName("release")
            proguardFiles(
                getDefaultProguardFile("proguard-android-optimize.txt"),
                "proguard-rules.pro"
            )
        }
    }
    compileOptions {
        sourceCompatibility = JavaVersion.VERSION_1_8
        targetCompatibility = JavaVersion.VERSION_1_8
    }
    kotlinOptions {
        jvmTarget = "1.8"
    }
    kotlin {
        jvmToolchain(8)
    }
    buildFeatures {
        compose = true
        viewBinding = true
        buildConfig = true
    }
    composeOptions {
        // https://developer.android.com/jetpack/androidx/releases/compose-kotlin
        kotlinCompilerExtensionVersion = "1.4.8"
    }
    packaging {
        resources {
            excludes += "/META-INF/{AL2.0,LGPL2.1}"
        }
    }
}

dependencies {
    implementation("build.buf:connect-kotlin-okhttp:0.1.7")
    implementation("build.buf:connect-kotlin-google-javalite-ext:0.1.7")
}

When I build the apk in release buildType with minify enabled. I got the following crash:

java.lang.IllegalArgumentException: No property for required constructor parameter #0 code of fun `<init>`(kotlin.String?, kotlin.String?, kotlin.collections.List<build.buf.connect.protocols.ErrorDetailPayloadJSON>?): build.buf.connect.protocols.ErrorPayloadJSON
	at com.squareup.moshi.kotlin.reflect.KotlinJsonAdapterFactory.create(KotlinJsonAdapter.kt:312)
	at com.squareup.moshi.ArrayJsonAdapter$1.create$bridge(ArrayJsonAdapter.java:0)
	at com.squareup.moshi.Moshi.adapter(Moshi.java:146)
	at com.squareup.moshi.Moshi.adapter(Moshi.java:106)
	at com.squareup.moshi.Moshi.adapter(Moshi.java:80)
	at build.buf.connect.protocols.ConnectInterceptor.parseConnectUnaryError(ConnectInterceptor.kt:201)
	at build.buf.connect.protocols.ConnectInterceptor.access$parseConnectUnaryError(ConnectInterceptor.kt:40)
	at build.buf.connect.protocols.ConnectInterceptor$unaryFunction$2.invoke(ConnectInterceptor.kt:91)
	at build.buf.connect.protocols.ConnectInterceptor$unaryFunction$2.invoke(ConnectInterceptor.kt:83)
	at build.buf.connect.protocols.ConnectInterceptor$unaryFunction$1.invoke$bridge(ConnectInterceptor.kt:0)
	at build.buf.connect.ProtocolClientConfig$chain$1$unaryFunction$2.invoke(ProtocolClientConfig.kt:129)
	at build.buf.connect.ProtocolClientConfig$chain$1$unaryFunction$2.invoke(ProtocolClientConfig.kt:126)

However, when I set isDebuggable = true, the crash disappeared.

I'm getting a crash calling this function, occasionally, it doesn't occur every time but it occurs about ⅓ of the time. It seems that if the function doesn't return the resultChannel.receive() in a timely manner, it hits the finally block and cancels the resultChannel, throwing the below exception. Also, shouldn't this function close the messageStream after receiving the result? I have to call the close() function before calling the receiveAndClose() function, otherwise my client doesn't receive from the result channel and my backend service does not process the resulting stream I sent to it. Thanks for looking into this!

FATAL EXCEPTION: OkHttp Dispatcher
    PID: 20742
    java.util.concurrent.CancellationException: Channel was cancelled
        at kotlinx.coroutines.channels.BufferedChannel.cancelImpl$kotlinx_coroutines_core(BufferedChannel.kt:1765)
        at kotlinx.coroutines.channels.BufferedChannel.cancel(BufferedChannel.kt:1762)
        at kotlinx.coroutines.channels.ReceiveChannel$DefaultImpls.cancel$default(Channel.kt:297)
        at build.buf.connect.impl.ClientOnlyStream.receiveAndClose(ClientOnlyStream.kt:36)
        at build.buf.connect.impl.ClientOnlyStream$receiveAndClose$1.invokeSuspend(Unknown Source:14)
        at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
        at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:106)
        at kotlinx.coroutines.internal.LimitedDispatcher$Worker.run(LimitedDispatcher.kt:115)
        at kotlinx.coroutines.scheduling.TaskImpl.run(Tasks.kt:100)
        at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:584)
        at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.executeTask(CoroutineScheduler.kt:793)
        at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.runWorker(CoroutineScheduler.kt:697)
        at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:684)

Here is my client code:

private suspend fun <PayloadType, RequestType, ResponseType> performNetworkCall(
        request: RequestType,
        performRequest: suspend (RequestType) -> ResponseMessage<ResponseType>,
        mapResponseOnSuccess: (ResponseMessage.Success<ResponseType>) -> PayloadType,
        enableLogs: Boolean = false
    ): Result<PayloadType> {
        if (enableLogs) {
            Logger.v("ConnectNetworkRequest: ${request.toString().redactPrivateFieldsConnectLibrary()}")
        }

        val isTokenValidResult = refreshTokenWithRetries()
        if (isTokenValidResult.isFailure) {
            return Result.failure(isTokenValidResult.exceptionOrNull() ?: Exception("Error token is no longer valid"))
        }

        return when (val response = performRequest(request)) {
            is ResponseMessage.Success -> {
                if (enableLogs) {
                    Logger.v("ConnectNetworkResponse: ${response.message.toString().redactPrivateFieldsConnectLibrary()}")
                }
                Result.success(mapResponseOnSuccess(response))
            }
            is ResponseMessage.Failure -> {
                Logger.e("ConnectNetworkResponse (Error): ${response.error}", response.error)
                Result.failure(response.error)
            }
        }
    }

override suspend fun savePhoto(
        file: File,
        imageType: String
    ): Result<String> {
        val imageInfo = ImageInfo
            .newBuilder()
            .setImageType(imageType)
            .build()

        return performNetworkCall(
            request = SavePhotoRequest
                .newBuilder()
                .setInfo(imageInfo)
                .build(),
            performRequest = { request ->
                val savePhoto = orderService.savePhoto()
                val response = savePhoto.send(request)
                val chunkData: MutableList<ByteString> = mutableListOf()
                file.forEachBlock(blockSize = AppConfig.getPhotoUploadChunkSize()) { buffer, bytesRead ->
                    Logger.v("ConnectNetworkRequest: savePhoto, bufferSize: ${buffer.size}, bytesRead: $bytesRead")
                    chunkData.add(ByteString.copyFrom(buffer, 0, bytesRead))
                }
                val response2: MutableList<Result<Unit>> = mutableListOf()
                chunkData.forEach { byteString ->
                    response2.add(
                        savePhoto.send(SavePhotoRequest
                            .newBuilder()
                            .setChunkData(byteString)
                            .build()
                        )
                    )
                }
                savePhoto.close()
                val result = savePhoto.receiveAndClose()
                if (response.isSuccess && response2.all { it.isSuccess }) {
                    ResponseMessage.Success(
                        message = result,
                        code = Code.OK,
                        headers = emptyMap(),
                        trailers = emptyMap()
                    )
                } else {
                    ResponseMessage.Failure(
                        error = ConnectError(code = Code.INTERNAL_ERROR),
                        code = Code.INTERNAL_ERROR,
                        headers = emptyMap(),
                        trailers = emptyMap()
                    )
                }
            },
            mapResponseOnSuccess = { "SUCCESS" }
        )
    }