how to pass reference asynchronously between threads? With Observer?

Question

How do I ensure that:

1.) localThread and remoteThread run independent of each other?

2.) pass messages between localThread and remoteThread?

Specifically, a String object from localThread needs to "percolate" up to Telnet through, I think it's known as, a call-back. However, there's not really anything, per se, for Telnet to observe. It's an anonymous reference to LocalIO, and I don't see that explicitly providing a reference helps.

I've read about java.util.concurrent.Semaphore until my head exploded, all I came away with was that it doesn't seem to apply. For these two threads, they should continue to run regardless of what the other thread is doing. However, there needs to be some mechanism to pass object references between the threads...

public class Telnet {

    public Telnet() throws InterruptedException {
        startThreads();
    }

    public static void main(String[] args) throws InterruptedException {
        new Telnet();
    }

    public void startThreads() throws InterruptedException {
        Semaphore s = new Semaphore(1, true);

        Thread localThread = new Thread(new LocalIO());
        Thread remoteThread = new Thread(new RemoteIO());

        localThread.start();
        remoteThread.start();
    }
}

The threads themselves are as follows. LocalIO:

public class LocalIO implements Runnable {

    @Override
    public void run() {
        Scanner scanner;
        String line;
        while (true) {
            scanner = new Scanner(System.in);
            line = scanner.nextLine();
            out.println("\n\nyou entered\t\"" + line + "\"\n");
        }
    }
}

RemoteIO:

public class RemoteIO implements Runnable {

    private static Logger log = Logger.getLogger(RemoteIO.class.getName());
    final String host = "rainmaker.wunderground.com";
    final int port = 3000;

    @Override
    public void run() {
        log.fine(host + port);
        int byteOfData;
        try (Socket socket = new Socket(host, port);
                InputStream inputStream = socket.getInputStream();
                OutputStream ouputStream = socket.getOutputStream();
                PrintWriter printWriter = new PrintWriter(socket.getOutputStream(), true);
                final BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(System.in))) {
            while ((byteOfData = inputStream.read()) != -1) {
                out.print((char) byteOfData);
            }
        } catch (Exception e) {
            out.println(e);
        }
    }
}

Keeping in mind that RemoteIO never closes its connection and runs indefinitely.

Answer 1

The solution I found:

public class RemoteConnection extends Observable {

    private static Logger log = Logger.getLogger(RemoteConnection.class.getName());
    private final Socket socket;
    private final BufferedInputStream in;
    private final BufferedOutputStream out;
    private final static String UTF8 = "UTF-8";

    public RemoteConnection(String host, int port) throws UnknownHostException, IOException {
        socket = new Socket(host, port);
        in = new BufferedInputStream(socket.getInputStream());
        out = new BufferedOutputStream(socket.getOutputStream());
    }

    public void write(Deque<String> commands) throws IOException {
        String command;
        while (!commands.isEmpty()) {
            command = commands.pop();
            out.write(command.concat("\r\n").getBytes(Charset.forName(UTF8)));
            log.info(command);
        }
        out.flush();
    }

    void read() {  //probably should use BufferedStream to better effect..?
        Thread readRemote = new Thread() {

            @Override
            public void run() {
                StringBuilder sb = new StringBuilder();
                char ch;
                int i;
                while (true) {
                    try {
                        i = in.read();
                        ch = (char) i;
                        sb.append(ch);
                        System.out.print(ch);
                        if (i == 13) {
                            setChanged();
                            notifyObservers(sb.toString());
                            log.fine(sb.toString());
                            sb = new StringBuilder();
                        }
                    } catch (IOException ioe) {
                        log.fine(ioe.toString());
                    }
                }
            }
        };
        readRemote.start();
    }
}

By reorganizing the threading, this approximates a poor-mans telnet, with asynchronous threads for i/o. I think reading from the console was blocking...something...

I really don't know why this works but the other approaches didn't. I would've preferred to have the main class start and handle threads, and pass references between the threads, but that just didn't work despite using the various solutions offered here.

LocalConnection has a similar threading approach.

Answer 2

The concurrent package is very helpful for this sort of thing: http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/package-summary.html

For example you can just give each thread a ConcurrentLinkedQueue and they can check the queue to see if there is anything to act on whenever they please. Meanwhile other threads can add new objects to the queue whenever they please.

Answer 3

There is one essential difference in the programming paradigm your code can adopt:

• synchronous mode: the receiving end runs an endless loop which explicitly takes items off a concurrent queue, blocking when there are no items ready;

• asynchronous mode: the receiving end submits a callback to an item exchange mechanism. This callback is called for every item which arrives from the producer thread.

The Observer pattern may loosely apply to the latter case, but not to the former.

Also note that in the latter case, the "item exchange mechanism" is usually implemented in the synchronous mode.

Answer 4

Not sure what yu are trying to do, but if you want to exchange data between threads, you need a volatile variable to make sure changes are seen by other threads. AtomicReferences are non-blocking and provide some API that might help here.