Java data object for bidirectional I/O

Question

I am developing an interface that takes as input an encrypted byte stream -- probably a very large one -- that generates output of more or less the same format.

The input format is this:

{N byte envelope}
    - encryption key IDs &c.
{X byte encrypted body}

The output format is the same.

Here's the usual use case (heavily pseudocoded, of course):

Message incomingMessage = new Message (inputStream);

ProcessingResults results = process (incomingMessage);

MessageEnvelope messageEnvelope = new MessageEnvelope ();
// set message encryption options &c. ...

Message outgoingMessage = new Message ();
outgoingMessage.setEnvelope (messageEnvelope);

writeProcessingResults (results, message);

message.writeToOutput (outputStream);

To me, it seems to make sense to use the same object to encapsulate this behaviour, but I'm at a bit of a loss as to how I should go about this. It isn't practical to load all of the encrypted body in at a time; I need to be able to stream it (so, I'll be using some kind of input stream filter to decrypt it) but at the same time I need to be able to write out new instances of this object. What's a good approach to making this work? What should Message look like internally?

Answer 1

Can you split the body at arbitrary locations?

If so, I would have two threads, input thread and output thread and have a concurrent queue of strings that the output thread monitors. Something like:

ConcurrentLinkedQueue<String> outputQueue = new ConcurrentLinkedQueue<String>();
...

private void readInput(Stream stream) {
    String str;
    while ((str = stream.readLine()) != null) {
       outputQueue.put(processStream(str));
    }
}

private String processStream(String input) {
    // do something
    return output;
}

private void writeOutput(Stream out) {
    while (true) {
        while (outputQueue.peek() == null) {
            sleep(100);
        }

        String msg = outputQueue.poll();
        out.write(msg);
    }
}

Note: This will definitely not work as-is. Just a suggestion of a design. Someone is welcome to edit this.

Answer 2

If you need to read and write same time you either have to use threads (different threads reading and writing) or asynchronous I/O (the java.nio package). Using input and output streams from different threads is not a problem.

If you want to make a streaming API in java, you should usually provide InputStream for reading and OutputStream for writing. This way those can then be passed for other APIs so that you can chain things and so get the streams go all the way as streams.

Input example:

Message message = new Message(inputStream);
results = process(message.getInputStream());

Output example:

Message message = new Message(outputStream);
writeContent(message.getOutputStream());

The message needs to wrap the given streams with a classes that do the needed encryption and decryption.

Note that reading multiple messages at same time or writing multiple messages at same time would need support from the protocol too. You need to get the synchronization correct.

Answer 3

You should check Wikipedia article on different block cipher modes supporting encryption of streams. The different encryption algorithms may support a subset of these.

Buffered streams will allow you to read, encrypt/decrypt and write in a loop.

Examples demonstrating ZipInputStream and ZipOutputStream could provide some guidance on how you may solve this. See example.

Answer 4

What you need is using Cipher Streams (CipherInputStream). Here is an example of how to use it.

Answer 5

I won't create one class to handle in- and output - one class, one responsibility. I would like two filter streams, one for input/decryption and one for output/encryption:

InputStream decrypted = new DecryptingStream(inputStream, decryptionParameters);
...
OutputStream encrypted = new EncryptingStream(outputSream, encryptionOptions);

They may have something like a lazy init mechanism reading the envelope before first read() call / writing the envelope before first write() call. You also use classes like Message or MessageEnvelope in the filter implementations, but they may stay package protected non API classes.

The processing will know nothing about de-/encryption just working on a stream. You may also use both streams for input and output at the same time during processing streaming the processing input and output.

Answer 6

I agree with Arne, the data processor shouldn't know about encryption, it just needs to read the decrypted body of the message, and write out the results, and stream filters should take care of encryption. However, since this is logically operating on the same piece of information (a Message), I think they should be packaged inside one class which handles the message format, although the encryption/decryption streams are indeed independent from this.

Here's my idea for the structure, flipping the architecture around somewhat, and moving the Message class outside the encryption streams:

class Message {
  InputStream input;
  Envelope envelope;

  public Message(InputStream input) {
    assert input != null;
    this.input = input;
  }

  public Message(Envelope envelope) {
    assert envelope != null;
    this.envelope = envelope;
  }

  public Envelope getEnvelope() {
    if (envelope == null && input != null) {
      // Read envelope from beginning of stream
      envelope = new Envelope(input);
    }
    return envelope
  }

  public InputStream read() {
    assert input != null

    // Initialise the decryption stream
    return new DecryptingStream(input, getEnvelope().getEncryptionParameters());
  }

  public OutputStream write(OutputStream output) {
    // Write envelope header to output stream
    getEnvelope().write(output);

    // Initialise the encryption
    return new EncryptingStream(output, getEnvelope().getEncryptionParameters());
  }
}

Now you can use it by creating a new message for the input, and one for the output: OutputStream output; // This is the stream for sending the message Message inputMessage = new Message(input); Message outputMessage = new Message(inputMessage.getEnvelope()); process(inputMessage.read(), outputMessage.write(output));

Now the process method just needs to read chunks of data as required from the input, and write results to the output:

public void process(InputStream input, OutputStream output) {
  byte[] buffer = new byte[1024];
  int read;
  while ((read = input.read(buffer) > 0) {
    // Process buffer, writing to output as you go.
  }
}

This all now works in lockstep, and you don't need any extra threads. You can also abort early without having to process the whole message (if the output stream is closed for example).