Using InputStream/OutputStream to handle data flows through modules

Question

so now I have a bunch of modules. It's like a assembly line. There is data flowing through and each module does something about the data, or you can say the next module consumes the output that the previous module produces. Each module is sort of expected to be standalone and reusable. This is quite a typical scenario I think.

So initially I have designed each module's interface as module(InputStream is, OutputStream os). So it can take files, whatever sources, as input and output. When you think of data flows, the previous module's outputstream is going to the next module's inputstream. But then I realise Java doesn't even have a intuitive/easy way to get data from OutputStream to InputStream. (Note: This question is not about how to achieve this. For those who is interested, How to convert OutputStream to InputStream?)

It seems to me that probably OutputStream/InputStream is not meant for purpose like this. So what is the best way to design the interface to handle data flows in this case?

Answer 1

For discrete objects you can use a producer/consumer pattern with ConcurrentLinkedQueues or BlockingQueues - each module has its own queue, and it will continually poll its queue (or use take if it's a BlockingQueue), process the object, and offer it to the next module's queue.

This pattern can also work with a byte stream if you chunk it into smaller byte arrays that you pass through the queues, but this isn't always appropriate (e.g. if module1 reads, module2 compresses, and module3 encrypts, then you're probably better off keeping the data in streams, unless you have some reasonable way to chunk the data).

Answer 2

You could create a general module which implements the OutputStream to InputStream conversion and place one instance of the module between each of your other modules. You could even get fancy with it and make the module intelligent enough to route the messages from one module to any of the other modules. This would become a sort of gateway or router type module.

Alternatively, you could implement something a bit heavier weight with a message queuing and passing framework like ZeroMQ.
--ap