How to notify a running thread of a change

Question

I am working on a client/server project which watches a system event in the server and sends a notification change to clients. It consists of a Watcher object which does the watching, and the server object which creates threads for each client connecting to it. Considering the simplified code below (I included all of the necessary code)

Now, the problem is that the notification is received from the mySrv object and the propertyChange of mySrv is executed properly and disMsg is populated. However, I don't know how to reach the run method of the ClientHandler to execute the notifyStatusChange() to send out the change to clients. I tried joining threads and many different things but none worked. I would really appreciate if someone knows a better approach or point me to my mistakes and best practices perhaps) or show me specifically how to achieve that.

Thanks,

class MyServer implements PropertyChangeListener {

    // all the fields and imports are removed for the sake of brevity 

    public class ClientHandler implements Runnable {
        BufferedReader reader;
        Socket socket;
        InputStreamReader iReader;

        public ClientHandler(Socket ClientSocket) {
            try {
                socket = ClientSocket;
                iReader = new InputStreamReader(socket.getInputStream());
                reader = new BufferedReader(iReader);
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }

        public void run() {
            String clientMessage;

            try {
                while (true) {
                    if (((clientMessage = reader.readLine()) != null)) {
                        System.out.println("Client: " + clientMessage);
                    }

                    if (isChanged) {
                        System.out.println("Server1: " + dispMsg);
                        notifyStatusChange();
                        isChanged = false;
                    }
                }
            } catch (IOException ex) {
                Logger.getLogger(MyServer.class.getName()).log(Level.SEVERE, null, ex);
            } catch (InterruptedException ex) {
                Logger.getLogger(MyServer.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
    }

    public void goServer() {

        try {
            ss = new ServerSocket(1234);

            while (true) {
                s = ss.accept();

                PrintWriter writer = new PrintWriter(s.getOutputStream(), true);
                Thread t = new Thread(new ClientHandler((s)), "Thread for Client #" + counter);
                t.start();

                System.out.println("Connection " + counter + " received from: "
                        + s.getInetAddress().getHostName());
                counter++;
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                ss.close();
            } catch (IOException ex) {
                Logger.getLogger(MyServer.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
    }

    public MyServer() {
        isChanged = false;
        if (dispMsg == null) {
            dispMsg = "Waiting for a change...";
        }
    }

    @Override
    public void propertyChange(PropertyChangeEvent pce) {

        isChanged = true;
        dispMsg = "Event to String: " + pce.toString() + "\n"
                + "Property Name: " + pce.getPropertyName() + "\n"
                + "Old Value" + pce.getOldValue() + "\n";

    }

    public void notifyStatusChange() throws IOException {
        PrintWriter pw = new PrintWriter(s.getOutputStream());
        pw.println(dispMsg);
    }

    public static void main(String[] args) {
        try {
            Watcher myWatcher = new WatcherDir();
            MyServer mySrv = new MyServer();

                myWatcher.addPropertyChangeListener(mySrv);
                myWatcher.start();
                mySrv.goServer();
                myWatcher.join();

        } catch (IOException ex) {
            Logger.getLogger(MyServer.class.getName()).log(Level.SEVERE, null, ex);
        } catch (InterruptedException ex) {
            Logger.getLogger(MyServer.class.getName()).log(Level.SEVERE, null, ex);
        }
    }
}

Answer 1

I can see a few different issues here.

It seems the server waits for a connection from a client and creates and starts ClientHandler thread when it receives a connection. Once such a connection is received, its run method will execute, in its own thread.

So at this point we have two threads, the server thread and the client thread, but if the client doesn't write anything to the socket after connecting then the thread will block at clientMessage = reader.readLine())

Now let us assume that propertyChange is called in the main thread, and it sets isChanged to true. If I understand right you are expecting notifyStatusChange to be called. But there are some problems:

1. ClientHandler.run may be blocked reading from client connection when isChanged becomes true

2. Even if something is read from the socket, the thread may find that isChanged is false. This is because the change occurred in a different thread. If the variable is declared to be volatile then this will not occur.

3. The server accepts multiple connections, with a different thread for each connection. But each thread's run method can set isChanged to false so even if the value is propagated to all threads, only one of them will print the notification.

I know this is not complete answer but you need to fix these problems.

Answer 2

A common pattern is producer-consumer. Threads can share a BlockingQueue[1], where producer thread adds items and consumer thread takes it from the collection.

There are many examples of implementations.

[1] http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/BlockingQueue.html

Answer 3

Is your "propertyChange()" getting executed, if so , you code should work without any issue. There is no need to execute "run()" again or join threads.

If you just want to notify a running Thread of a change, the following simple code will work.

package indika.java.core.network;



public class MyServer{
public boolean isChanged = false;
private String dispMsg;

public MyServer() throws InterruptedException {
    Thread mthread = new Thread(new ClientThread());
    mthread.start();
    Thread.sleep(1000);
    isChanged = true;
}

public class ClientThread implements Runnable {

    public void run() {
        boolean ok = true;
        while (ok) {
            System.out.println("Started thread!!!!");
            if (isChanged) {
                System.out.println("Server1: " + dispMsg);
                notifyStatusChange();
                isChanged = false;
                ok = false;
            }
        }
    }
}

public void notifyStatusChange() {
    System.out.println("notifyStatusChange -- ");
}

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