/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements. See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership. The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License. You may obtain a copy of the License at
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied. See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package org.apache.stratos.messaging.message.receiver.instance.notifier;


import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.stratos.messaging.broker.subscribe.EventSubscriber;
import org.apache.stratos.messaging.listener.EventListener;
import org.apache.stratos.messaging.util.MessagingUtil;

/**
 * A thread for receiving instance notifier information from message broker.
 */
public class InstanceNotifierEventReceiver {
    private static final Log log = LogFactory.getLog(InstanceNotifierEventReceiver.class);
    private final InstanceNotifierEventMessageDelegator messageDelegator;
    private final InstanceNotifierEventMessageListener messageListener;
    private EventSubscriber eventSubscriber;
    private boolean terminated;

    public InstanceNotifierEventReceiver() {
        InstanceNotifierEventMessageQueue messageQueue = new InstanceNotifierEventMessageQueue();
        this.messageDelegator = new InstanceNotifierEventMessageDelegator(messageQueue);
        this.messageListener = new InstanceNotifierEventMessageListener(messageQueue);
    }

    public void addEventListener(EventListener eventListener) {
        messageDelegator.addEventListener(eventListener);
    }


    public void execute() {
        try {
            // Start topic subscriber thread
            eventSubscriber = new EventSubscriber(MessagingUtil.Topics.INSTANCE_NOTIFIER_TOPIC.getTopicName(), messageListener);
//			subscriber.setMessageListener(messageListener);
            Thread subscriberThread = new Thread(eventSubscriber);

            subscriberThread.start();
            if (log.isDebugEnabled()) {
                log.debug("InstanceNotifier event message receiver thread started");
            }

            // Start instance notifier event message delegator thread
            Thread receiverThread = new Thread(messageDelegator);
            receiverThread.start();
            if (log.isDebugEnabled()) {
                log.debug("InstanceNotifier event message delegator thread started");
            }

            // Keep the thread live until terminated
            while (!terminated) {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException ignore) {
                }
            }
        } catch (Exception e) {
            if (log.isErrorEnabled()) {
                log.error("InstanceNotifier receiver failed", e);
            }
        }
    }

    public boolean isSubscribed() {
        return ((eventSubscriber != null) && (eventSubscriber.isSubscribed()));
    }

    public void terminate() {
        eventSubscriber.terminate();
        messageDelegator.terminate();
        terminated = true;
    }
}

@@ -872,17 +872,16 @@
 ifier;%0A%0A
-%0A
 import o
@@ -1443,80 +1443,8 @@
 or;%0A
-    private final InstanceNotifierEventMessageListener messageListener;%0A
     
@@ -1753,29 +1753,61 @@
 e);%0A        
-this.
+InstanceNotifierEventMessageListener 
 messageListe
@@ -1864,24 +1864,206 @@
 sageQueue);%0A
+        // Start topic subscriber thread%0A        eventSubscriber = new EventSubscriber(MessagingUtil.Topics.INSTANCE_NOTIFIER_TOPIC.getTopicName(),%0A                messageListener);%0A
     %7D%0A%0A    p
@@ -2186,17 +2186,16 @@
 %0A    %7D%0A%0A
-%0A
     publ
@@ -2226,99 +2226,144 @@
     
-try %7B%0A            // Start topic subscriber thread%0A            eventSubscriber = new
+synchronized (this) %7B%0A            if (terminated) %7B%0A                log.info(%22InstanceNotifierEventReceiver has been terminated.
  Event
-S
+ s
 ubsc
@@ -2371,141 +2371,93 @@
 iber
-(MessagingUtil.Topics.INSTANCE_NOTIFIER_TOPIC.getTopicName(), messageListener);%0A//%09%09%09subscriber.setMessageListener(messageListener);%0A
+ will not be created.%22);%0A                return;%0A            %7D%0A            try %7B%0A    
     
@@ -2519,17 +2519,20 @@
 riber);%0A
-%0A
+    
         
@@ -2565,32 +2565,36 @@
 ();%0A            
+    
 if (log.isDebugE
@@ -2613,32 +2613,36 @@
                 
+   
+ 
 log.debug(%22Insta
@@ -2698,32 +2698,36 @@
 %22);%0A            
+    
 %7D%0A%0A            /
@@ -2717,32 +2717,36 @@
  %7D%0A%0A            
+ 
+   
 // Start instanc
@@ -2795,24 +2795,28 @@
             
+  
+  
 Thread recei
@@ -2853,24 +2853,28 @@
 Delegator);%0A
+    
             
@@ -2901,32 +2901,36 @@
 ();%0A            
+    
 if (log.isDebugE
@@ -2949,32 +2949,36 @@
                 
+  
+  
 log.debug(%22Insta
@@ -3035,34 +3035,37 @@
 %22);%0A            
+    
 %7D
-%0A
 %0A            // 
@@ -3065,104 +3065,145 @@
     
-// Keep the thread live until terminated%0A            while (!terminated) %7B%0A                try %7B
+%7D catch (Exception e) %7B%0A                if (log.isErrorEnabled()) %7B%0A                    log.error(%22InstanceNotifier receiver failed%22, e);
 %0A   
@@ -3219,40 +3219,22 @@
     
+%7D%0A
     
-Thread.sleep(1000);%0A
         
     
@@ -3221,32 +3221,34 @@
   %7D%0A            
+%7D%0A
         %7D catch 
@@ -3244,216 +3244,253 @@
    %7D
- catch (InterruptedException ignore) %7B%0A                %7D%0A            %7D%0A        %7D catch (Exception e) %7B%0A            if (log.isErrorEnabled()) %7B%0A                log.error(%22InstanceNotifier receiver failed%22, e);
+%0A        log.info(%22InstanceNotifierEventReceiver started%22);%0A%0A        // Keep the thread live until terminated%0A        while (!terminated) %7B%0A            try %7B%0A                Thread.sleep(2000);%0A            %7D catch (InterruptedException ignore) %7B
 %0A   
@@ -3647,24 +3647,37 @@
 %0A    public 
+synchronized 
 void termina
@@ -3785,16 +3785,78 @@
 = true;%0A
+        log.info(%22InstanceNotifierEventReceiver terminated%22);%0A
     %7D%0A%7D%0A

Add synchronize blocks to InstanceNotifierEventReceiver to avoid concurrency issues