2
\$\begingroup\$

For a university project I'm trying to build a system that has a general architecture of:

Client app <=====> Client library < - - > Server library <=====> Server app

I'm trying to make the client and server as general (in the context of a java application, cross-platform compatibility is a concern for later) as possible, so the client library serializing objects from the client app to send to the server and deserializing objects from the server as responses.

I have the client library written, and a simple test application that just sends String objects to the client library for sending to the server. I've gotten cleanup code which (I believe) works fairly well in the client library, but the cleanup code in the test app doesn't seem to work, even though it's quite similar. It leaves its threads running and doesn't shut down. The TestApp.UpstreamThread remains in the running state, while the TestApp.DownstreamThread remains in the waiting state forever. As you can see from the code, when one thread ends it's supposed to interrupt() the other.

The code for the test app is as follows:

public class TestApp {

    private UpstreamChannel     upstream            = null;
    private DownstreamChannel   downstream          = null;
    private Thread              upstreamThread      = null;
    private Thread              downstreamThread    = null;
    private Client <String>     client              = null;
    private boolean             ending              = false;

    /**
     * 
     */
    private class UpstreamChannel implements Runnable {

        private TestApp outer   = null;

        /**
         * 
         */
        @Override
        public void run () {

            Thread.currentThread ().setName ("TestApp.UpstreamChannel");

            try (BufferedReader inReader = new BufferedReader (new InputStreamReader (System.in))) {
                while (!this.outer.ending) {
                    this.outer.client.acceptMessage (inReader.readLine ());
                }
            } catch (IOException ex) {
                Logger.getLogger (this.getClass ().getName ()).log (Level.SEVERE, null, ex);
            } finally {
                this.outer.ending   = true;
                this.outer.downstreamThread.interrupt ();
                Thread.currentThread ().interrupt ();
                return;
            }
        }

        /**
         * 
         * @param app 
         */
        public UpstreamChannel (TestApp app) {
            this.outer  = app;
        }
    }

    /**
     * 
     */
    private class DownstreamChannel implements Runnable {

        private TestApp outer   = null;

        /**
         * 
         */
        @Override
        public void run () {

            Thread.currentThread ().setName ("TestApp.DownstreamChannel");

            try {
                while (!this.outer.ending) {
                    System.out.println (this.outer.client.getMessage ());
                }
            } catch (InterruptedException ex) {
                Logger.getLogger (this.getClass ().getName ()).log (Level.INFO, null, ex);
            } finally {
                this.outer.ending   = true;
                this.outer.upstreamThread.interrupt ();
                Thread.currentThread ().interrupt ();
                return;
            }
        }

        /**
         * 
         * @param app 
         */
        public DownstreamChannel (TestApp app) {
            this.outer  = app;
        }
    }

    /**
     * 
     */
    public void run () {
        if ((null == this.upstreamThread) 
        && (null == this.downstreamThread)) {
            this.upstreamThread     = new Thread (this.upstream);
            this.downstreamThread   = new Thread (this.downstream);
            this.upstreamThread.start ();
            this.downstreamThread.start ();
        }
    }

    /**
     * 
     */
    public TestApp (Client <String> client) {
        this.upstream       = new UpstreamChannel (this);
        this.downstream     = new DownstreamChannel (this);
        this.client         = client;

        Logger.getLogger (this.getClass ().getName ()).log (Level.INFO, "Class instantiated");
    }
}

And this is the client library:

public class Client <T> {
    private String              hostname            = "localhost";
    private Integer             port                = 8113;
    private Socket              connection          = null;
    private DownstreamChannel   downstream          = null;
    private UpstreamChannel     upstream            = null;
    private Thread              upstreamThread      = null;
    private Thread              downstreamThread    = null;

    /**
     * This thread handles waiting for messages from the attached app and 
     * sending them to the server
     */
    private class UpstreamChannel <T> implements Runnable {

        private Client              outer       = null;
        private BlockingQueue <T>   inputQueue  = null;

        /**
         * Send messages to server
         * 
         * This method waits for input from the attached app and writes it to 
         * the server's socket as a serialized object.  
         */
        @Override
        public void run () {

            Thread.currentThread ().setName ("Client.UpstreamChannel");

            try (ObjectOutputStream outWriter = new ObjectOutputStream (this.outer.connection.getOutputStream ())) {
                while (true) {
                    // Get last input line from the user
                    outWriter.writeObject (this.inputQueue.take ());
                    outWriter.flush ();
                }
            } catch (IOException | InterruptedException ex) {
                Logger.getLogger (this.getClass ().getName ()).log (Level.SEVERE, null, ex);
            } finally {
                this.outer.downstreamThread.interrupt ();
                return;
            }
        }

        /**
         * Accept a message to send upstream
         * 
         * @param message
         * @return 
         */
        public UpstreamChannel acceptMessage (T message) {
            this.inputQueue.add (message);
            return this;
        }

        /**
         * Initialize the upstream channel
         * 
         * @param outer
         */
        public UpstreamChannel (Client outer) {
            this.outer      = outer;
            this.inputQueue = new LinkedBlockingQueue ();
        }
    }

    /**
     * This thread handles waiting for messages from the server and sending them
     * to the connected client
     */
    private class DownstreamChannel <T> implements Runnable {

        private Client              outer       = null;
        private BlockingQueue <T>   outputQueue = null;

        /**
         * Get messages from server
         * 
         * This loop waits for messages from the server and pushes them into
         * the downstream queue
         */
        @Override
        public void run () {

            Thread.currentThread ().setName ("Client.DownstreamChannel");

            try (ObjectInputStream inReader = new ObjectInputStream (this.outer.connection.getInputStream ())) {
                while (true) {
                    this.outputQueue.add ((T) inReader.readObject ());
                }
            } catch (IOException | ClassNotFoundException ex) {
                Logger.getLogger (this.getClass().getName()).log(Level.SEVERE, null, ex);
            } finally {
                this.outer.upstreamThread.interrupt ();
                return;
            }
        }

        /**
         * Get the first unprocessed message from the downstream queue
         * 
         * @return
         * @throws InterruptedException 
         */
        public T getMessage () throws InterruptedException {
            return this.outputQueue.take ();
        }

        /**
         * Initialize the downstream channel
         * 
         * @param outer
         */
        public DownstreamChannel (Client outer) {
            this.outer          = outer;
            this.outputQueue    = new LinkedBlockingQueue ();
        }
    }

    /**
     * Start the main loop
     */
    public void run () throws IOException, InterruptedException {
        if ((null == this.upstreamThread) 
        && (null == this.downstreamThread)) {
            this.upstreamThread     = new Thread (this.upstream);
            this.downstreamThread   = new Thread (this.downstream);

            this.upstreamThread.start ();
            this.downstreamThread.start ();
        }
    }

    /**
     * Send a message to the server
     * 
     * @param message
     * @return 
     */
    public Client acceptMessage (T message) {
        this.upstream.acceptMessage (message);
        return this;
    }

    /**
     * Get the next unprocessed message from the server
     * 
     * @return
     * @throws InterruptedException 
     */
    public T getMessage () throws InterruptedException {
        return (T) this.downstream.getMessage ();
    }

    /**
     * Initialize the client
     * 
     * @param hostname
     * @param port 
     */
    public Client (String hostname, Integer port) throws UnknownHostException, IOException {
        this.hostname   = hostname;
        this.port       = port;
        this.connection = new Socket (this.hostname, this.port);
        this.upstream   = new UpstreamChannel (this);
        this.downstream = new DownstreamChannel (this);
    }
}

The class that runs the code is basically just a stub but I've included it for completeness.

public class TestClient {

    /**
     * 
     * @param args
     * @throws Exception 
     */
    public static void main (String[] args) throws UnknownHostException, IOException, InterruptedException {
        Client <String> clientInstance  = new Client ("localhost", 8113);
        TestApp         app             = new TestApp (clientInstance);

        clientInstance.run ();
        app.run ();
    }
}

Whilst the test app is obviously a fairly disposable bit of code, I would like it to clean up properly as it will form the skeleton on which the actual client applications will be built.

I'm also curious as to the quality of the code I've produced (which I think this is better suited to a code review than a question on Stack Exchange). I've been trying to follow recommendations as regards cleaning up after yourself, and it does seem to work, but I'm wondering if it's sufficiently robust. Any input on the cleanup code quality would be appreciated.

\$\endgroup\$
  • \$\begingroup\$ I suspect the issue is one of the client library needing to notify the application that it's ceased talking to the server. I just need to figure out a decent mechanism for doing it. \$\endgroup\$ – GordonM Aug 1 '13 at 10:11
  • \$\begingroup\$ Currently you have no code or interface that even tries to initiate a clean shutdown. The Socket is never closed. \$\endgroup\$ – bowmore Aug 2 '13 at 16:06
  • \$\begingroup\$ It's a try-with-resource block. try (ObjectOutputStream outWriter = new ObjectOutputStream (this.outer.connection.getOutputStream ())) means that Java should automatically close the inputstreamreader when the try block exits so you don't have to put it there explicitly. It's a java 7 feature. \$\endgroup\$ – GordonM Aug 2 '13 at 20:16
  • \$\begingroup\$ I know, the point is : nothing ever triggers it to close. \$\endgroup\$ – bowmore Aug 2 '13 at 22:30
2
\$\begingroup\$

Tidbits for you:

private UpstreamChannel     upstream            = null;
private boolean             ending              = false;

Instance variables are initialized to null by default, and boolean values are set to false by default.

Without applying self-encapsulation, the classes will not be as extensible as they could be.

The following lines:

    Logger.getLogger (this.getClass ().getName ()).log (Level.SEVERE, null, ex);
    Logger.getLogger (this.getClass ().getName ()).log (Level.INFO, "Class instantiated");

Have duplicate code that could be eliminated using:

    getLogger().log( ... );

Then:

private void Logger getLogger() {
  return Logger.getLogger( this.getClass ().getName () );
}

This method can be eliminated altogether using a static LOGGER instance variable and obtaining the class name dynamically. And it can be simplified even more by using aspects to inject logging when required.

An empty return line from the finally block is superfluous.

Avoid code such as:

            this.outer.ending   = true;

Instead:

            getOuter().endDownstream();

The endDownstream() method would resemble:

public void end() {
  setEnding( true );
}

public void endDownstream() {
  end();
  getDownstreamThread().interrupt();
}

The corresponding endUpstream() method:

public void enUpstream() {
  end();
  getUpstreamThread().interrupt();
}

This can be refactored further, to merge both methods into a single call (note that the end() method can be eliminated, as well, since setEnding(true) now only exists once in the code):

public void endStream() {
  setEnding( true );
  getStreamThread().interrupt();
}

There is a bit of duplication between UpstreamChannel and DownstreamChannel. The duplication can be abstracted using a common AbstractChannel class via polymorphism. For example, it appears as though all channels have a name, which leads to duplicate code such as:

        Thread.currentThread ().setName ("Client.UpstreamChannel");

This can be written, instead, as:

public class AbstractChannel <T> implements Runnable {
  public void run() {
    Thread.currentThread().setName (getName());
  }

  protected abstract String getName();
}

private class UpstreamChannel <T> extends AbstractChannel {
    protected String getName() {
      return "Client.UpstreamChannel";
    }

    public void run () {
        super.run();
        // ...

Now the superclass isolates how the name is set in a single location for all subclasses. The subclasses simply change getName() to return their name.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.