Using PipedInputStream and PipedOutputStream with Threads

[Layne Lund]

I am working on a project to encrypt some data using XOR encryption. (Yes, I know XOR is not a strong encryption scheme, but I'm just getting my feet wet in cryptography and using this as a stepping stone to more complicated things.) Anyways, I wrote a class called Xor that looks like the following:

public class Xor {
    private byte[] key;
 
    public Xor(byte[] key) {
        this.key = key;
    }
 
    public void encrypt(InputStream in, OutputStream out) throws IOException {
        //...
    }
}

The encrypt() method takes characters from the given InputStream and writes encrypted data to the OutputStream. Now, I also wrote a helper class called StreamManipulator that does some other transformations on a given InputStream and writes to a given OutputStream:

public class StreamManipulator {
    public void removeWhitespace(InputStream in, OutputStream out) throws IOException {
        //...
    }
 
    public void toLowerCase(InputStream in, OutputStream out) throws IOException {
        //...
    }
}

As the name indicate, removeWhitespace() reads characters from the given InputStream and writes all non-whitespace characters to the given OutputStream. The toLowerCase() method reads characters from the given InputStream, converts them to lower-case, and writes the converted characters to the given OutputStream.

I feel like this is a good design since each method does a small task. Alternatively, Xor.encrypt() could do all the tasks at once, but the logic in that one method would become monstrous. Also, if I create a Encryptor interface, I can easily replace the XOR cipher with a different one. (This will probably happen down the road at some point.)

Now, I want to put these all together: take a InputStream (at the moment this will be a FileInputStream, but I'm coding to the interface not the implementation ) and convert all the characters to lower case, remove all whitespace, then encrypt it using the XOR cipher. One solution is to create temporary files and write each intermediate step to disk. However, I would like to avoid the overhead of file I/O that this would create. I also don't want to read the whole stream into memory as a String and process it from there. (Hence the decision to take stream paramters instead of Strings.) So instead, I read about PipedInputStream and PipedOutputStream. These seem like a good solution to my problem as long as I can come up with a decent scheme to create the threads for each step in the process. However, I have come up against a few problems, mostly regarding IOExceptions. Let me illustrate with the code that I have come up with so far:

    private static void encrypt(final InputStream in, final OutputStream out) {
        final PipedOutputStream lowerCaseOut = new PipedOutputStream();
        final PipedInputStream lowerCaseIn = new PipedInputStream(lowerCaseOut);
        final PipedOutputStream noWhitespaceOut = new PipedOutputStream();
        final PipedInputStream noWhitespaceIn = new PipedInputStream(noWhitespaceOut);
        
        new Thread(new Runnable() {
            public void run() {
                try {
                    new StreamManipulator().toLowerCase(in, lowerCaseOut);
                } catch (IOException ex) {
                    System.err.println("Error occured while converting to lower case:");
                    ex.printStackTrace();
                } finally {
                    in.close();
                    lowerCaseOut.close();
                }
            }
        }).run();
        
        new Thread(new Runnable() {
            public void run() {
                try {
                    new StreamManipulator().removeWhitespace(lowerCaseIn, noWhitespaceOut);
                } catch (IOException ex) {
                    System.err.println("Error occured while removing whitespace:");
                    ex.printStackTrace();
                } finally {
                    lowerCaseIn.close();
                    noWhitespaceOut.close();
                }
            }
        }).run();
        
        new Thread(new Runnable() {
            public void run() {
                Random rand = new Random();
                byte[] key = new byte[rand.nextInt(Integer.MAX_VALUE)];
                rand.nextBytes(key);
                
                try {
                    new Xor(key).encrypt(noWhitespaceIn, out);
                } catch (IOException ex) {
                    System.err.println("Error occured while encrypting data:");
                    ex.printStackTrace();
                } finally {
                    noWhitespaceIn.close();
                    out.close();
                }
            }
        }).run();
    }

The first problem it doesn't compile since close() throws an IOException. So my first question is how do I deal with this cleanly? Do I add another try...catch statement to my finally clauses? This seems a little ugly to me, so is there a better way to do this? Even more importantly, am I even approaching this in the right way? I don't know a lot about piped streams, so maybe I need to back up and figure them out better. But this thread is already long enough, so I'll start another thread when I want to ask more general questions about the behavior of PipedInputStream and PipedOutputStream.

Well, that's it for now. I would appreciate any feedback or comments anyone has.

Regards,

Layne

[Stan James]

Here's a different approach. A filter reads an input and writes an output on its own thread. Piped streams connect multiple filters. You could customize the behavior of the filters by plugging in strategies, little classes that just do one manipulation each.

package com.saxman.sandbox.pipedstreams;
 
import java.io.*;
 
public class PipeFilter implements Runnable
{
    private InputStream in = null;
    private OutputStream out = null;
    private static final int BUFFER_SIZE = 1024;
    private byte[] buffer = new byte[BUFFER_SIZE];
    private int length = 0;
 
    public PipeFilter(InputStream is, OutputStream os)
    {
        in = is;
        out = os;
    }
 
    public void run()
    {
        try
        {
            while (-1 != (length = in.read( buffer )))
            {
                massageBuffer();
                out.write( buffer, 0, length );
            }
        }
        catch (Exception e)
        {
            e.printStackTrace();
        }
        closeStreams();
    }
 
    private void massageBuffer()
    {
        System.out.println( in.toString() + " " + new String( buffer, 0, length ) );
    }
 
    public void closeStreams()
    {
        try
        {
            in.close();
        }
        catch (Exception e)
        {
            e.printStackTrace();
        }
        try
        {
            out.close();
        }
        catch (Exception e)
        {
            e.printStackTrace();
        }
    }
}
= = = = = = = = = = = = = = = =
package com.saxman.sandbox.pipedstreams;
 
import java.io.*;
 
import junit.framework.TestCase;
import junit.textui.TestRunner;
 
public class PipeTester extends TestCase
{
    public static void main(String[] args)
    {
        TestRunner.run( PipeTester.class );
    }
 
    public void testOne() throws Exception 
    {
        ByteArrayInputStream source = new ByteArrayInputStream( "Hello".getBytes() );
        ByteArrayOutputStream target = new ByteArrayOutputStream();
        Thread pf1 = new Thread( new PipeFilter( source, target ) );
        pf1.start();
        pf1.join();
        assertEquals( "Hello", target.toString() );
    }
 
    public void testTwo() throws Exception
    {
        ByteArrayInputStream source = new ByteArrayInputStream( "Hello".getBytes() );
        PipedOutputStream pipeOut = new PipedOutputStream();
        PipedInputStream pipeIn = new PipedInputStream( pipeOut );
        ByteArrayOutputStream target = new ByteArrayOutputStream();
        Thread pf1 = new Thread( new PipeFilter( source, pipeOut ) );
        Thread pf2 = new Thread( new PipeFilter( pipeIn, target ) );
        pf1.start();
        pf2.start();
        pf1.join();
        pf2.join();
        assertEquals( "Hello", target.toString() );
    }
}

I did something like this to send objects through the pipeline instead of bytes using BlockingQueues instead of piped streams. It wasn't quite fast enough for MIDI message manipulation.

[Layne Lund]

Thanks for the code snippet. In my research on this, I found the FilterInputStream, FilterOutputStream, FilterReader, and FilterWriter classes. At least for screening out whitespace characters and converting to lower case, I think I will extend FilterReader. That way I can easily use the Decorator pattern to nest different readers and only read the file once for all the "filtering" I want to do.

As for the encryption part, I'm debating whether to extend FilterOutputStream. However, there doesn't seem to be any benefit to outweigh the extra complexity that I will need to add to my implementation.

Layne

[Layne Lund]

Okay, I decided to extend FilterReader to read the characters and filter out the non-alphabetic characters and convert them to lower-case. This reduces the amount of code needed in the driver since I just create several nested Readers and call read() on the outer-most one. I'm still writing the intermediate filtered data to a temporary file, which I don't really like, but it will work for now. I'm ready to go onto the real meat of this project.

Layne