Pipeline `Transform` concept

[wkornewald]

Currently, transformations like encrypt/decrypt and compress/decompress have to be implemented as Source/Sink. If you want to support encryption you have to implement it twice as EncryptionSource and EncryptionSink because the decision where the encryption is applied depends on how you want to structure the code from a security and abstraction perspective. In Node.js the Streams API supports such a Transform concept already.

What we see with Okio's GzipSink and GzipSource is that you can only decompress a source and compress a sink. That's a little bit inflexible though because you have to write your code around this limitation. Written as a pipeline it's clear that compression is its own concept, independent of sink and source.

source | compress | sink
source | decompress | sink

It would be nice to introduce a general Transform API with source.pipe(transform) -> Source and transform.pipe(sink) -> Sink operations. See also Node.js as an example, but not necessarily for how we should structure the API: https://nodejs.org/api/stream.html#class-streamtransform

This is just an example how this could work (but please let's first agree on the need for a Transform concept): A Transform could, for example, receive an input and output Buffer and then be allowed to use UnsafeBufferOperations on both to transfer data from the source to the sink. It would also need some way to request a minimum amount of data to be present in the input Buffer and only be called again when the Buffer has the requested size or the input has been closed / has reached the end of the stream (in which case the Transform can process the remaining data or fail).

[JakeWharton]

That's a little bit inflexible though because you have to write your code around this limitation.

Do you have concrete examples of this being a problem in practice? In general transformations tend to bias themselves in one direction or the other.

For example, I have a hard time seeing the use of reading the first 100 bytes of the gzip compressed form of an underlying text file (having read 1000 bytes of it). But I absolutely see the utility in reading only the first 100 bytes of the decompressed form of a gzip file (having read only 50 bytes of it).

Written as a pipeline it's clear that compression is its own concept, independent of sink and source.

I think this is a bit hand-wavy on details. Who is driving the advancement of the data in this world and where are the intermediate buffers?

Your chart can just as easily be written as

decompress(source) | sink
source | compress(sink)

and still make sense, and that's the current state of things.

In a shell with pipes there's a whole process with its own threads and intermediate buffers on both sides of that compress/decompress step. In-process we don't want to pay for both when we only need one side. Hence why the operations are biased towards the direction against which they realistically only make sense.

[wkornewald]

One example would be to apply encryption on the source, so the rest of the code can't get access to the unencrypted data and is free to decide whether to transfer over the network or write to a file etc. In unit tests we also often do something like source.pipe(encrypt).readByteArray() and make assertions against the result directly without extra boilerplate for piping into a sink (we already have a Transform API internally and find it more convenient than hard-coding against Sink/Source).

Also, there is no logical reason why encryption should only be applicable on a sink. Encryption is logically independent of source and sink. It's just a data transformation, not a provider/consumer of data on its own. I don't understand why the API should be designed around a bias instead of being designed around the true nature of the concept.

Regarding efficiency: As suggested above, the Transform can work with UnsafeBufferOperations to make the whole pipeline zero-copy (that's how our implementation works). You take the input Buffer's blocks with UnsafeBufferOperations and pass them to cipher.update() and put the result into the output Buffer via UnsafeBufferOperations. If the next pipeline step is Base64 encoding that can again do the same. No copies at all. And surely one can build a nicer API around this.

[kpavlov]

Reading from source, transforming, and consuming are common patterns indeed. I think #487 can address it

[JakeWharton]

Also, there is no logical reason why encryption should only be applicable on a sink.

Encrypting or compressing a (raw) source is really no different than the existing support for buffering a source. It's a decorator which provides a hidden, intermediate sink (i.e., buffer) into which data can be hydrated in response to a read.

I think you could support this today by just offering a transformation lambda of the backing Buffer, something like rawSource.buffered { it.gzipped() }.

I don't understand why the API should be designed around a bias instead of being designed around the true nature of the concept.

The reason that each direction of encryption, compression, etc. was biased towards source and sink is that it avoids those intermediate buffers where they not necessary. When you decorate a sink with gzip or a source with (un)gzip, the segments are immediately passed along to the next stage of the pipeline when available.

I think that fact actually makes me like a .buffered { }-style source transformer more, because it emphasizes in the API the fact that in order to accomplish that shape of layering there is an additional intermediate buffering layer being applied as opposed to simply applying it directly to a sink.

[wkornewald]

With the extra buffering, do you mean the internal ByteArray segments being ownership-moved (not copied) from one Buffer to the other?

In Okio, with a GzipSource the original Source gets buffered and the GzipSource transforms from that source Buffer directly into the sink Buffer. So, we have two Buffers (plus maybe the original source's buffer, but that could in theory be optimized away).

With Transform, if you do source.pipe(GzipDecompressor()) you'd operate on one source Buffer and the result could be a RawSource which can then be piped into a (Raw)Sink which means one more Buffer. That's again two Buffers. Alternatively, if we made the result type a Source then we'd have another Buffer, but when piping into a Sink that can easily be optimized to skip the first pipe's Buffer and process directly into the final Sink's Buffer (so there are three Buffers in memory, but only two are used for the pipeline).

So, I'm still not sure what you mean. Could you please provide an example where a Transform is unable to achieve the same amount of buffering as Source/Sink biased wrappers?