A modern programming language that makes developers happier.

kotlinlang

Kind of an open ended question but how are folks handling backpressure in relation to coroutines?
Say I have an existing pipeline in project reactor, messages are pumped in one side, queued, fed into a ParallelFlux and then transformed and ultimately consumed by the app. The queue and the "rails" in the pflux effectively give you back pressure so only X messages are in flight in the pipeline concurrently and as one message on a rail finishes it picks up the next from the queue.
That seems like it'd be pretty hard to model as easily with coroutines. Am I missing something?

It is “given” with coroutines “automagically”. You don’t even have to think about backpressure with coroutines. All coroutines communication primitives (channels) automatically suspend producers when consumer don’t keep up.

I dont think that's true, `kotlinx.coroutines.channels.*` expose a bunch of backpressure controls which all default to _pay it forward_. The rendezvous channel for example will suspend `send`ers (producers) after the first one of too many elements has been produced.

I don't know enough about reactor to know how you send things to it, but, assuming it exposes some kind of non-blocking callback-based API to let you know when its accepted a message, you could write your own adapter:

```
val inputReactorPipelineObject: Reactor.PipelineEntrance&lt;T&gt; = //...

val inputChannel = actor&lt;T&gt; { 
  consumeEach { message: T -&gt; 
    suspendCoroutineOrReturn&lt;Unit&gt; { continuation -&gt; 
      inputReactorPipelineObject.offer(message, onMessageProcessingStarted = { continuation.resume(Unit) })
      COROUTINE_SUSPENDED;
    }
  }
}
```

if not, and it simply has an `offer` method that blocks until the message processing starts, then it presumably has an executor it lets you configure, if thats the case...

```
val reactorExecutor: Executor = //...
val inpuReactorPipelineObject: Reactor.PipelineEntrance&lt;T&gt; = ...

val inputChannel: SendChannel&lt;T&gt; = actor&lt;T&gt;(context = reactorExecutor.asCoroutineDispatcher()) {
  consumeEach { message -&gt; 
    //call the blocking API and block, but since we're on the executor used by the message, its as good as we're gonna get.
    inpuReactorPipelineObject.offer(message)
  } 
}
```

a key question then becomes one of exception handling, and that's handled by channel _failure states_, which act as you might expect: if `offer` throws some kind of exception, that exception will get thrown at the caller of `inputChannel.send()`

... _it might be wraped in a `ClosedChannelException`... I cant remember_...

Thanks. I'll take a closer look around these areas now I have an idea of where to start and see if I can model what I already have in reactor.