How to manually concatenate the arguments of a function call?

[AHardworkingMan]

Hi folks!
After I added the tool to the ChatCompletionCreateParameters, it seems that it only returns a complete function call parameter once.
I use the following code to make the backend server forward the streaming results.

StreamResponse<ChatCompletionChunk> streamResponse = client.chat().completions().createStreaming(params);
Observable<ChatCompletionChunk> observable = Observable.fromIterable(streamResponse.stream()::Iterator);
Flowable<ChatCompletionChunk> flowable = observable.toFlowable(BackpressureStrategy.BUFFER);
Flowable<ServerSentEvent> sse = flowable.map(chunk->{
    handleGenerationResult(chunk);
    return ServerSentEvent.builder().id(String.valueOf(eventCount).event("ai-response").data(nowContent.toString()).build();
});
Publisher<ServerSentEvent> s = transformable(sse);
ctx.render(ServerSentEvents.builder().build(s));

I want it to output some reasons for the call before the tool is invoked, but it seems to only output the tokens from the first response.
If the params does not contain the tool, the streaming response will proceed as normal.
If I make a second repeated request, it will be returned as a stream.
Is it possible to transmit the tool_call in a streaming manner? I can concatenate the parameters myself.

Thanks a lot! 😀

[AHardworkingMan]

Ah, this should be my mistake, unrelated to the SDK. I didn't manually close the stream. After changing it to the method below, there were no issues.

private Flowable<ServerSentEvent> streamCall(ChatCompletionCreatePatams params){
  return Flowable.create(emitter -> {
      try(StreamResponse<ChatCompletionChunk> streamResponse = client.chat().completions().createStreaming(params)){
          streamResponse.stream().forEach(chunk -> {
              try{
                  handleGenerationResult(chunk);
                  emitter.onNext(
                      ServerSentEvent.builder()
                          .id(String.valueOf(eventCount)).event("ai-response").data(nowContent.toString()).build();
                  );
              }catch(JsonProcessingException e){
                  throw new RuntimeException(e);
              }
          });
      }
      emitter.onComplete();
  }, BackpressureStrategy.BUFFER);
}