softwaremill / softwaremill/sttp

Remove fs2 interop with reactive-streams

Open
#2,963 2 comments 2 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

Dominant language
Scala
Stars
1.5k
Forks
338
Avg merge
8h 59m
Merged PRs (30d)
13

Description

I have recently discovered that HttpClientFs2Backend uses interop with reactive-streams in order to support streaming capabilities. This has several issues:

  • Firstly, this interop doesn't support cancellation: https://github.com/typelevel/fs2/issues/3260
  • Secondly, implementation can use direct interop between java.util.concurrent.Flow and fs2, and avoid extra layer of interop with reactive streams.

Since sttp requires JDK 11, using native java.util.concurrent.Flow seems to be more correct

For more context, I would also like to provide a task in which issue was discovered. I was implementing a handling of SSE stream of events in the background of my application, and I wanted to add an idle timeout (abort stream if there are no events for some period of time) using fs2 instrumentation timeoutOnPullTo:

  • In case of using HttpClientFs2Backend, cancellation never finished, connection stayed open and moreover application never closed
  • In case of using Http4sBackend, everything worked properly

In order to reproduce, I forked the HttpClientFs2Backend and changed implementation of lowLevelBodyToBody to use fs2.Stream.fromPublisher . This resolved an issue for me, now cancellation worked properly. Thus, I am creating this issue.

Finally, I believe the same behaviour can be obtained with ArmeriaFs2Backend because the same interop of fs2 with reactive-streams is used. However, I do not know what to do there, because Armeria uses reative-streams in the API

Contributor guide

No contributing guide indexed for this repository

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

Research direction

Start with HttpClientFs2Backend.lowLevelBodyToBody and compare its reactive-streams interop with fs2.Stream.fromPublisher; also inspect ArmeriaFs2Backend, which may use the same approach. Reproduce the SSE idle-timeout scenario with fs2 timeoutOnPullTo, and consider the work complete when cancellation closes the stream and application without leaving the connection open.

Written by the indexing model from the issue text.

Assessment

Tech stack
java, scala
Domain
networking
Issue type
Refactor
Difficulty
4/5
Estimated time
3-5 days
Activity status
Active
Clarity
Mostly clear
Newbie friendliness
48/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.