Registering the Observer in Obervable

[s ravi chandran]

Hi,

I have a problem. I have a main application, which starts two services on two different thread. Service 1 is observer and Service 2 is observable. When Service 1 starts, I pass its instance to a singleton class. When Service 2 starts, I use reference of Service 1 from this singleton class and fire events.

Problem starts with the sequence of how Service 1 and Service 2 starts. On my local machine, the Services run in sequence. But on a server, it is going haywire. Service 2 and Service 1 run in parallel and by the time Service 2 is up, Service 1 is not available. And when Service 1 comes up, the fired events are already lost.

What is the best possible way to control it? Is use of Blocking Queue a possible solution?

[Stephan van Hulst]

Hard to say without knowing more about what your application is supposed to do.

Regardless, using mutable global data is a very bad idea, especially so in a multi-threaded environment. Don't use singletons.

[Rajdeep Biswas]

Yup, please explain more. From the little understanding that I got, I think both threads are not doing inter-thread communication properly, or probably you needed to implement some polling mechanism against some state of your services. If you are knowing what you are trying to accomplish, BlockingQueue might be a good option too.

[s ravi chandran]

Rajdeep Biswas wrote:Yup, please explain more. From the little understanding that I got, I think both threads are not doing inter-thread communication properly, or probably you needed to implement some polling mechanism against some state of your services. If you are knowing what you are trying to accomplish, BlockingQueue might be a good option too.

Yes. You are right. Till now there was no need for these two services to communicate. Now the situation has changed. And because I do not want to bind their working into any sequence, I have not added any thread controls. Service 1 does a type of processing, Service 2 does another type of processing. Requirement is that when Service 2 does a processing, it has to trigger an event in Service 1. This will initiate the processing at Service 1. But because these services can start processing without waiting for each other, there are complications.

Let me try to create a codebase reflecting this scenario. That should help in explaining a bit.

[s ravi chandran]

This is what I think represents my current setup. Let me know if any component is not clear.

public class Service2 {

    private EventDispatcher dispatcher;

    public Service2()
    {
        // initialize the service

        // get dispatcher instance
        dispatcher = EventDispatcher.getInstance();
    }

    public void start() {
        System.out.println("Service 2 started");
    }

    // event trigger based on other conditions
    public void triggerService2EventProcessing(Object event1)
    {
        System.out.println("Service 2 processing event");
        dispatcher.dispatchEvent(event1);
    }

}

public class Service1 implements Observer{

    private Map<String, String> xmlTransformMap;

   public Service1()
    {
        // initialize Service1
        init();

        // pass instance to event dispatcher
        EventDispatcher.getInstance().setObserver(this);
    }

    public void init()
    {
        xmlTransformMap = new HashMap<>();
        // populate xmlTransformMap;
    }

    public void start() {

        System.out.println("Service 1 started");
    }

    @Override
    public void update(Observable o, Object event) {
        // transform the event and process it here
        triggerService1EventProcessing();
    }

    private void triggerService1EventProcessing() {

        // use xmlTransformMap to calculate values for event
        xmlTransformMap.get("eventkey");
        System.out.println("Service 1 processing event");
    }
}

public class EventDispatcher extends Observable{

    private static EventDispatcher instance;

    private EventDispatcher()
    {

    }

    public void setObserver(Service1 observer) {
        addObserver(observer);
    }

    public void dispatchEvent(Object event)
    {
        setChanged();

        notifyObservers(event);
    }

    public static EventDispatcher getInstance()
    {
        if(instance==null)
        {
            instance = new EventDispatcher();
        }

        return instance;
    }

}

public class ApplicationLauncher {

    public static void main(String[] args) {

        new Thread(
                new Runnable(){
                    public void run()
                    {
                        new Service1();
                    }
                }
        ).start();

        new Thread(
                new Runnable(){
                    public void run()
                    {
                        new Service2();
                    }
                }
        ).start();

    }
}

Here if Service 2 comes up before Service 1, it will start processing events and sending it to Event Dispatcher. Event Dispatcher will update the Service 1 if it is registered by that time. Else the events are lost.

I want a solution for this issue.

[gyank kannur]

I guess this should be moved to the Threads forum!

[Stephan van Hulst]

This is really just a producer-consumer problem. You can pass messages through a BlockingQueue implementation. Note that you shouldn't be creating your own threads. Use an ExecutorService instead.

import java.util.concurrent.*;

final class Sender implements Runnable {
 
  private final BlockingQueue<Message> messagePipe;
 
  Sender(BlockingQueue<Message> messagePipe) {
    if (messagePipe == null)
      throw new IllegalArgumentException();
 
    this.messagePipe = messagePipe;
  }
 
  @Override
  public void run() {
    try {
      for (int i = 0; i < 100; i++) {
        Message message = new Message(i);
        messagePipe.put(message);
      }
    }
    
    catch (InterruptedException ex) {
      Thread.currentThread().interrupt();
    }
  }
}

import java.util.concurrent.*;

final class Receiver implements Runnable {
 
  private final BlockingQueue<Message> messagePipe;
 
  Receiver(BlockingQueue<Message> messagePipe) {
    if (messagePipe == null)
      throw new IllegalArgumentException();
 
    this.messagePipe = messagePipe;
  }
 
  @Override
  public void run() {
    try {
      for (int i = 0; i < 100; i++) {
        Message message = messagePipe.take();
        System.out.println(message);
      }
    }
    
    catch (InterruptedException ex) {
      Thread.currentThread().interrupt();
    }
  }
}

import java.util.concurrent.*;

final class Application {
 
  private final ExecutorService        executor    = Executors.newCachedThreadPool();
  private final BlockingQueue<Message> messagePipe = new LinkedBlockingQueue<>();
  private final Sender                 sender      = new Sender  (messagePipe);
  private final Receiver               receiver    = new Receiver(messagePipe);
 
  void run() {
    if (executor.isShutdown())
      throw new IllegalStateException();
 
    try {
      executor.execute(sender  );
      executor.execute(receiver);
    }
 
    finally {
      executor.shutdown();
    }
  }
 
  public static void main(String... args) {
    new Application().run();
  }
}

[s ravi chandran]

Thread creation is not under my control. I just wanted to represent that once the threads are spawn, its initialization and execution is independent of each other. A third party library takes the class names and creates the objects on-the-fly using reflection. It is only after the creation, during initialization that I get to use those. I do think my current design is way too complex. Hence my wish to use blocking queue to simply this.

[Stephan van Hulst]

Yes, BlockingQueue is just fine for this problem. Just don't store it in a global variable. Pass it to your services' constructors.

[Tim Moores]

Observer and Observable go back to the early days of Java; I wouldn't use them for anything at this point. A more modern approach would be to use an event bus, like the one that's part of Guava: https://github.com/google/guava/wiki/EventBusExplained. It removes the need to pass instances of your event objects to all concerned pieces of code.

[Mike Simmons]

Tim Moores wrote:Observer and Observable go back to the early days of Java; I wouldn't use them for anything at this point.

Well, java.util.Observer and java.util.Observable go back to the early days of Java. They didn't really catch on and are not generally used anymore. However the observer/observable pattern is still around, and may be experiencing a renaissance in places like ReactiveX. It's not clear to me that the poster is talking about the old java.util implementation, or something else.

[Tim Moores]

Right, the pattern is still valid. The code specifically uses those classes, though, so I wanted to caution against that.

[s ravi chandran]

I am indeed using the Java Util Observer and Observable implementations. Will check out the Event Bus. If we have this library available to us, will use this instead.

[s ravi chandran]

Planning to use this class in place of the observer pattern. Have tried running a simulation and found it working as expected. Only this, I need to use it as a singleton, as I do not have access to the object creation, I need to inject it to the services.

public class MessageBroker {

    BlockingQueue<Integer> broker;
    private static final MessageBroker INSTANCE = new MessageBroker();

    public MessageBroker()
    {
        broker = new LinkedBlockingQueue<>();
    }

    public void send(Integer value) {
        try {
            broker.put(value);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public Integer receive() {
        try {
            return broker.take();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return null;
    }

    public static MessageBroker getInstance()
    {
        return INSTANCE;
    }

}

[Stephan van Hulst]

If you're using an application container, you can usually inject your classes without having to resort to static variables.

What framework are you using?

[s ravi chandran]

Stephan van Hulst wrote:If you're using an application container, you can usually inject your classes without having to resort to static variables.

What framework are you using?

It is a custom third party library being used. We have a properties file where we define the classes that we will run and it uses reflection to instantiate them. Once they are initialized, we do our processing. The library doesn't have option to pass params to the instantiating class.