path: root/tags/java/sca/1.5.1-RC4/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/NotificationTypeManagerImpl.java

/*
 * 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.tuscany.sca.binding.notification;

import java.io.InputStream;
import java.io.OutputStream;
import java.net.URI;
import java.net.URL;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import javax.servlet.ServletInputStream;
import javax.servlet.ServletOutputStream;

import org.apache.tuscany.sca.binding.notification.encoding.Broker;
import org.apache.tuscany.sca.binding.notification.encoding.BrokerConsumerReference;
import org.apache.tuscany.sca.binding.notification.encoding.BrokerID;
import org.apache.tuscany.sca.binding.notification.encoding.BrokerProducerReference;
import org.apache.tuscany.sca.binding.notification.encoding.Brokers;
import org.apache.tuscany.sca.binding.notification.encoding.Constants;
import org.apache.tuscany.sca.binding.notification.encoding.EncodingException;
import org.apache.tuscany.sca.binding.notification.encoding.EncodingObject;
import org.apache.tuscany.sca.binding.notification.encoding.EncodingRegistry;
import org.apache.tuscany.sca.binding.notification.encoding.EncodingUtils;
import org.apache.tuscany.sca.binding.notification.encoding.EndConsumers;
import org.apache.tuscany.sca.binding.notification.encoding.EndProducers;
import org.apache.tuscany.sca.binding.notification.encoding.EndpointAddress;
import org.apache.tuscany.sca.binding.notification.encoding.EndpointReference;
import org.apache.tuscany.sca.binding.notification.encoding.EndpointReferenceWrapper;
import org.apache.tuscany.sca.binding.notification.encoding.NeighborBrokerConsumers;
import org.apache.tuscany.sca.binding.notification.encoding.Neighbors;
import org.apache.tuscany.sca.binding.notification.encoding.NewBroker;
import org.apache.tuscany.sca.binding.notification.encoding.NewBrokerAck;
import org.apache.tuscany.sca.binding.notification.encoding.NewBrokerResponse;
import org.apache.tuscany.sca.binding.notification.encoding.NewConsumer;
import org.apache.tuscany.sca.binding.notification.encoding.NewConsumerResponse;
import org.apache.tuscany.sca.binding.notification.encoding.NewProducer;
import org.apache.tuscany.sca.binding.notification.encoding.NewProducerResponse;
import org.apache.tuscany.sca.binding.notification.encoding.RemoveBroker;
import org.apache.tuscany.sca.binding.notification.encoding.RemovedBroker;
import org.apache.tuscany.sca.binding.notification.encoding.ReplaceBrokerConnection;
import org.apache.tuscany.sca.binding.notification.util.IOUtils;
import org.apache.tuscany.sca.binding.notification.util.NotificationServlet;
import org.apache.tuscany.sca.binding.notification.util.URIUtil;
import org.apache.tuscany.sca.binding.notification.util.IOUtils.IOUtilsException;
import org.apache.tuscany.sca.binding.notification.util.IOUtils.ReadableContinuation;
import org.apache.tuscany.sca.binding.notification.util.IOUtils.Writeable;
import org.apache.tuscany.sca.binding.notification.util.NotificationServlet.NotificationServletStreamHandler;
import org.apache.tuscany.sca.host.http.ServletHost;

/**
 * A notification type manager serves as a registry of producers and consumers, or brokers, for
 * any notification type. This class implements an interface that allows a reference provider
 * (a producer), a service provider (a consumer), or both (a broker, via the provider factory),
 * to access locally the ntm for its notification type, regardless of whether the ntm resides
 * locally or remotely.
 * At a given host there is only one reference provider and/or one service provider for any given
 * notification type. So, if the ntm for a notification type resides locally, then it is invoked
 * exclusively by either a reference provider (newProducer), a service provider (newConsumer), or
 * a provider factory (newBroker). And since these invocations occur when the providers are being
 * created then all three of consumerLists, producerLists and brokerLists must be null when these
 * invocations occur.
 * 
 * @version $Rev$ $Date$
 */
public class NotificationTypeManagerImpl implements NotificationTypeManager {

    private static final String ntmPathBase = "/ntm";

    private ServletHost servletHost;
    private EncodingRegistry encodingRegistry;
    private Map<URI, NotificationTypeManagerHandler> ntmHandlers;
    
    public NotificationTypeManagerImpl() {
    }

    public void setServletHost(ServletHost servletHost) {
        this.servletHost = servletHost;
    }

    public void setEncodingRegistry(EncodingRegistry encodingRegistry) {
        this.encodingRegistry = encodingRegistry;
    }

    public void init() {
        ntmHandlers = new HashMap<URI, NotificationTypeManagerHandler>();
    }
    
    public String newConsumer(URI notificationType, URL consumerUrl, URL remoteNtmUrl, List<URL> producerListResult) {
        if (ntmUrlIsRemote(consumerUrl, remoteNtmUrl)) {
            try {
                WriteableEPW wEPW = new WriteableEPW(new NewConsumer(), consumerUrl);
                InputStreamDecoder isd = new InputStreamDecoder();
                NewConsumerResponse ncr =
                    (NewConsumerResponse)IOUtils.sendHttpRequest(remoteNtmUrl, Constants.NEW_CONSUMER_OP, wEPW, isd);
                String sequenceType = ncr.getSequenceType();
                if (Constants.EndProducers.equals(sequenceType) || Constants.BrokerProducers.equals(sequenceType)) {
                    for (EndpointReference epr : ncr.getReferenceSequence()) {
                        producerListResult.add(epr.getEndpointAddress().getAddress());
                    }
                }
                return sequenceType;
            } catch(Exception e) {
                e.printStackTrace();
                throw new RuntimeException(e);
            }
        } else {
            NotificationTypeManagerHandler ntmHandler = ntmHandlers.get(notificationType);
            if (ntmHandler != null) {
                throw new RuntimeException("Trying to deploy local consumer with existing local producer, consumer or broker");
            }
            
            createNtmHandler(consumerUrl.getAuthority(), notificationType, consumerUrl, null, null);
            
            return Constants.NoProducers;
        }
    }
    
    private void createNtmHandler(String ntmUriAuthority, URI notificationType, URL consumerUrl, URL producerUrl, BrokerStruct broker) {
        String ntmUri = "http://" + ntmUriAuthority + ntmPathBase + URIUtil.getPath(notificationType);
        NotificationTypeManagerHandler ntmh = new NotificationTypeManagerHandler(notificationType, consumerUrl, producerUrl, broker);
        ntmHandlers.put(notificationType, ntmh);
        servletHost.addServletMapping(ntmUri, new NotificationServlet(ntmh));
    }
    
    public String newProducer(URI notificationType, URL producerUrl, URL remoteNtmUrl, List<URL> consumerListResult) {
        if (ntmUrlIsRemote(producerUrl, remoteNtmUrl)) {
            try {
                WriteableEPW wEPW = new WriteableEPW(new NewProducer(), producerUrl);
                InputStreamDecoder isd = new InputStreamDecoder();
                NewProducerResponse npr =
                    (NewProducerResponse)IOUtils.sendHttpRequest(remoteNtmUrl, Constants.NEW_PRODUCER_OP, wEPW, isd);
                String sequenceType = npr.getSequenceType();
                if (Constants.EndConsumers.equals(sequenceType) || Constants.BrokerConsumers.equals(sequenceType)) {
                    for (EndpointReference epr : npr.getReferenceSequence()) {
                        consumerListResult.add(epr.getEndpointAddress().getAddress());
                    }
                }
                return sequenceType;
            } catch(Exception e) {
                e.printStackTrace();
                throw new RuntimeException(e);
            }
        } else {
            NotificationTypeManagerHandler ntmHandler = ntmHandlers.get(notificationType);
            if (ntmHandler != null) {
                throw new RuntimeException("Trying to deploy local producer with existing local producer, consumer or broker");
            }
            
            createNtmHandler(producerUrl.getAuthority(), notificationType, null, producerUrl, null);
            
            return Constants.NoConsumers;
        }
    }
    
    public boolean newBroker(URI notificationType,
                             URL consumerUrl,
                             URL producerUrl,
                             String brokerID,
                             URL remoteNtmUrl,
                             List<EndpointReference> consumerListResult,
                             List<EndpointReference> producerListResult) {
        String ntmUriAuthority = producerUrl.getAuthority();
        if (!ntmUriAuthority.equals(consumerUrl.getAuthority())) {
            throw new RuntimeException("Producer url and consumer url do not match");
        }
        if (ntmUrlIsRemote(producerUrl, remoteNtmUrl)) {
            try {
                WriteableNewBroker wnb = new WriteableNewBroker(consumerUrl, producerUrl, brokerID);
                InputStreamDecoder isd = new InputStreamDecoder();
                NewBrokerResponse nbr =
                    (NewBrokerResponse)IOUtils.sendHttpRequest(remoteNtmUrl, Constants.NEW_BROKER_OP, wnb, isd);
                if (nbr.isFirstBroker()) {
                    if (nbr.getEndConsumers().getSequenceType().equals(Constants.EndConsumers)) {
                        for (EndpointReference epr : nbr.getEndConsumers().getReferenceSequence()) {
                            consumerListResult.add(epr);
                        }
                    }
                    if (nbr.getEndProducers().getSequenceType().equals(Constants.EndProducers)) {
                        for (EndpointReference epr : nbr.getEndProducers().getReferenceSequence()) {
                            producerListResult.add(epr);
                        }
                    }
                } else {
                    for (Broker broker : nbr.getBrokers().getBrokerSequence()) {
                        consumerListResult.add(broker.getBrokerConsumerReference().getReference());
                        producerListResult.add(broker.getBrokerProducerReference().getReference());
                    }
                }
                return nbr.isFirstBroker();
            } catch(Exception e) {
                e.printStackTrace();
                throw new RuntimeException(e);
            }
        } else {
            NotificationTypeManagerHandler ntmHandler = ntmHandlers.get(notificationType);
            if (ntmHandler != null) {
                throw new RuntimeException("Trying to deploy local broker with existing local producer, consumer or broker");
            }
            
            BrokerStruct broker = new BrokerStruct(consumerUrl, producerUrl, brokerID);
            createNtmHandler(ntmUriAuthority, notificationType, null, null, broker);
            
            return true;
        }
    }
    
    private boolean ntmUrlIsRemote(URL localUrl, URL ntmUrl) {
        if (ntmUrl == null) {
            return false;
        }
        if (localUrl.getPort() != ntmUrl.getPort()) {
            return true;
        }
        String remoteNtmUrlAuthority = ntmUrl.getAuthority();
        if (remoteNtmUrlAuthority.indexOf("localhost") >= 0) {
            return false;
        }
        return !localUrl.getAuthority().equals(remoteNtmUrlAuthority);
    }
    
    public void newBrokerAck(URL remoteNtmUrl) {
        try {
            IOUtils.sendHttpRequest(remoteNtmUrl, Constants.NEW_BROKER_ACK_OP, new WriteableNewBrokerAck(), null);
        } catch(Exception e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }
    
    public void removeBroker(EndpointReference brokerConsumerEpr, List<EndpointReference> neighborBrokerConsumerEprs, URL remoteNtmUrl) {
        WriteableRemoveBroker wrb = new WriteableRemoveBroker(brokerConsumerEpr, neighborBrokerConsumerEprs);
        
        try {
            IOUtils.sendHttpRequest(remoteNtmUrl, Constants.REMOVE_BROKER_OP, wrb, null);
        } catch(Exception e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }
    
    private class NotificationTypeManagerHandler implements NotificationServletStreamHandler {
        
        private URI notificationType;
        List<URL> consumerList;
        List<URL> producerList;
        List<BrokerStruct> brokerList;
        private NotificationTypeLock notificationTypeLock;
        private BrokerStruct pendingBroker;
        
        public NotificationTypeManagerHandler(URI notificationType) {
            this.notificationType = notificationType;
            this.notificationTypeLock = new NotificationTypeLock();
            this.pendingBroker = null;
        }
        
        public NotificationTypeManagerHandler(URI notificationType, URL consumerUrl, URL producerUrl, BrokerStruct broker) {
            this(notificationType);
            if (consumerUrl != null) {
                addConsumer(consumerUrl);
            } else if (producerUrl != null) {
                addProducer(producerUrl);
            } else if (broker != null) {
                addBroker(broker);
            }
        }
        
        private void addConsumer(URL consumerUrl) {
            if (consumerList == null) {
                consumerList = new ArrayList<URL>();
            }
            consumerList.add(consumerUrl);
        }
        
        private void addProducer(URL producerUrl) {
            if (producerList == null) {
                producerList = new ArrayList<URL>();
            }
            producerList.add(producerUrl);
        }
        
        private void addBroker(BrokerStruct broker) {
            if (brokerList == null) {
                brokerList = new ArrayList<BrokerStruct>();
            }
            brokerList.add(broker);
        }
        
        public void handle(Map<String, String> headers, ServletInputStream istream, int contentLength, ServletOutputStream ostream) {
            String opHeader = headers.get(IOUtils.Notification_Operation);
            EncodingObject eo = null;
            try {
                eo = EncodingUtils.decodeFromStream(encodingRegistry, istream);
            } catch(EncodingException e) {
                throw new RuntimeException(e);
            }
            
            if (Constants.NEW_CONSUMER_OP.equals(opHeader)) {
                handleNewConsumer((NewConsumer)eo, ostream);
            } else if(Constants.NEW_PRODUCER_OP.equals(opHeader)) {
                handleNewProducer((NewProducer)eo, ostream);
            } else if(Constants.NEW_BROKER_OP.equals(opHeader)) {
                handleNewBroker((NewBroker)eo, ostream);
            } else if (Constants.NEW_BROKER_ACK_OP.equals(opHeader)) {
                handleNewBrokerAck();
            } else if (Constants.REMOVE_BROKER_OP.equals(opHeader)) {
                handleRemoveBroker((RemoveBroker)eo);
            }
        }
        
        private void handleNewConsumer(NewConsumer nc, ServletOutputStream ostream) {
            synchronized(notificationTypeLock) {
                if (notificationTypeLock.isLocked) {
                    try { notificationTypeLock.wait(); } catch(InterruptedException e) {}
                }
                URL consumerUrl = nc.getReference().getEndpointAddress().getAddress();
                if (brokerList == null) {
                    addConsumer(consumerUrl);
                }
    
                NewConsumerResponse ncr = new NewConsumerResponse();
                if (producerList != null) {
                    ncr.setSequenceType(Constants.EndProducers);
                    for (URL producerUrl : producerList) {
                        ncr.addReferenceToSequence(EncodingUtils.createEndpointReference(producerUrl, null));
                    }
                } else if(brokerList != null) {
                    ncr.setSequenceType(Constants.BrokerProducers);
                    for (BrokerStruct broker : brokerList) {
                        ncr.addReferenceToSequence(EncodingUtils.createEndpointReference(broker.producerUrl, null));
                    }
                } else {
                    ncr.setSequenceType(Constants.NoProducers);
                }
                try {
                    EncodingUtils.encodeToStream(encodingRegistry, ncr, ostream);
                } catch(IOUtilsException e) {
                    throw new RuntimeException(e);
                }
            }
        }
        
        private void handleNewProducer(NewProducer np, ServletOutputStream ostream) {
            synchronized(notificationTypeLock) {
                if (notificationTypeLock.isLocked) {
                    try { notificationTypeLock.wait(); } catch(InterruptedException e) {}
                }
                URL producerUrl = np.getReference().getEndpointAddress().getAddress();
                if (brokerList == null) {
                    addProducer(producerUrl);
                }
    
                NewProducerResponse npr = new NewProducerResponse();
                if (consumerList != null) {
                    npr.setSequenceType(Constants.EndConsumers);
                    for (URL consumerUrl : consumerList) {
                        npr.addReferenceToSequence(EncodingUtils.createEndpointReference(consumerUrl, null));
                    }
                } else if(brokerList != null) {
                    npr.setSequenceType(Constants.BrokerConsumers);
                    for (BrokerStruct broker : brokerList) {
                        npr.addReferenceToSequence(EncodingUtils.createEndpointReference(broker.consumerUrl, null));
                    }
                } else {
                    npr.setSequenceType(Constants.NoConsumers);
                }
                try {
                    EncodingUtils.encodeToStream(encodingRegistry, npr, ostream);
                } catch(IOUtilsException e) {
                    throw new RuntimeException(e);
                }
            }
        }
        
        private void handleNewBroker(NewBroker nb, ServletOutputStream ostream) {
            synchronized(notificationTypeLock) {
                if (notificationTypeLock.isLocked) {
                    try { notificationTypeLock.wait(); } catch(InterruptedException e) {}
                }
               NewBrokerResponse nbr = new NewBrokerResponse();
                if (consumerList != null || producerList != null || brokerList == null) {
                    nbr.setFirstBroker(true);
                    EndConsumers endConsumers = new EndConsumers();
                    if (consumerList != null) {
                        endConsumers.setSequenceType(Constants.EndConsumers);
                        for (URL consumerUrl : consumerList) {
                            endConsumers.addReferenceToSequence(EncodingUtils.createEndpointReference(consumerUrl, null));
                        }
                    } else {
                        endConsumers.setSequenceType(Constants.NoConsumers);
                    }
                    nbr.setEndConsumers(endConsumers);
                    EndProducers endProducers = new EndProducers();
                    if (producerList != null) {
                        endProducers.setSequenceType(Constants.EndProducers);
                        for (URL producerUrl : producerList) {
                            endProducers.addReferenceToSequence(EncodingUtils.createEndpointReference(producerUrl, null));
                        }
                    } else {
                        endProducers.setSequenceType(Constants.NoProducers);
                    }
                    nbr.setEndProducers(endProducers);
                } else {
                    nbr.setFirstBroker(false);
                    Brokers brokers = new Brokers();
                    for (BrokerStruct brokerStruct : brokerList) {
                        Broker brokerElt = new Broker();
                        BrokerConsumerReference bcr = new BrokerConsumerReference();
                        bcr.setReference(EncodingUtils.createEndpointReference(brokerStruct.consumerUrl, brokerStruct.brokerID));
                        brokerElt.setBrokerConsumerReference(bcr);
                            
                        BrokerProducerReference bpr = new BrokerProducerReference();
                        bpr.setReference(EncodingUtils.createEndpointReference(brokerStruct.producerUrl, brokerStruct.brokerID));
                        brokerElt.setBrokerProducerReference(bpr);
                        brokers.addBrokerToSequence(brokerElt);
                    }
                    nbr.setBrokers(brokers);
                }
                EndpointReference consumerEPR = nb.getBrokerConsumerReference().getReference();
                URL consumerUrl = consumerEPR.getEndpointAddress().getAddress();
                BrokerID consumerBrokerID = consumerEPR.getReferenceProperties().getProperty(BrokerID.class);
                EndpointReference producerEPR = nb.getBrokerProducerReference().getReference(); 
                URL producerUrl = producerEPR.getEndpointAddress().getAddress();
                BrokerID producerBrokerID = producerEPR.getReferenceProperties().getProperty(BrokerID.class);
                if (consumerBrokerID == null ||
                        producerBrokerID == null ||
                            !consumerBrokerID.getID().equals(producerBrokerID.getID())) {
                    throw new RuntimeException("Producer and consumer broker ids do not match");
                }
                // only add broker if consumerList == null && producerList == null
                // otherwise, make it a pending broker and wait for ack
                // TODO block for a configurable amount of time
                BrokerStruct broker = new BrokerStruct(consumerUrl, producerUrl, consumerBrokerID.getID());
                if (consumerList == null && producerList == null) {
                    addBroker(broker);
                } else {
                    pendingBroker = broker;
                    notificationTypeLock.isLocked = true;
                }
                try {
                    EncodingUtils.encodeToStream(encodingRegistry, nbr, ostream);
                } catch(IOUtilsException e) {
                    throw new RuntimeException(e);
                }
            }
        }
        
        private void handleNewBrokerAck() {
            synchronized(notificationTypeLock) {
                if (!notificationTypeLock.isLocked) {
                    notificationTypeLock.notifyAll();
                    throw new RuntimeException("Notification type should be locked");
                }
                if (brokerList != null) {
                    notificationTypeLock.isLocked = false;
                    notificationTypeLock.notifyAll();
                    throw new RuntimeException("Can't add pending broker to non-empty broker list");
                }
                if (pendingBroker == null) {
                    notificationTypeLock.isLocked = false;
                    notificationTypeLock.notifyAll();
                    throw new RuntimeException("Missing pending broker");
                }
                addBroker(pendingBroker);
                consumerList = null;
                producerList = null;
                pendingBroker = null;
                notificationTypeLock.isLocked = false;
                notificationTypeLock.notifyAll();
            }
        }
        
        private void handleRemoveBroker(RemoveBroker rb) {
            synchronized(notificationTypeLock) {
                if (notificationTypeLock.isLocked) {
                    try { notificationTypeLock.wait(); } catch(InterruptedException e) {}
                }
                
                if (brokerList == null) {
                    throw new RuntimeException("No broker to remove for [" + notificationType + "]");
                }
                
                NeighborBrokerConsumers nbcs = rb.getNeighborBrokerConsumers();
                EndpointReference rbEpr = rb.getBrokerConsumerReference().getReference();
                if (nbcs != null && nbcs.getReferenceSequence() != null) {
                    List<Broker> neighborBrokers = new ArrayList<Broker>();
                    for (EndpointReference neighborBrokerConsumerEpr : nbcs.getReferenceSequence()) {
                        BrokerStruct neighborBrokerStruct = null;
                        URL neighborBrokerConsumerEprUrl = neighborBrokerConsumerEpr.getEndpointAddress().getAddress();
                        for (BrokerStruct brokerStruct : brokerList) {
                            if (brokerStruct.consumerUrl.equals(neighborBrokerConsumerEprUrl)) {
                                neighborBrokerStruct = brokerStruct;
                                break;
                            }
                        }
                        if (neighborBrokerStruct == null) {
                            throw new RuntimeException("Can't find neighbor broker for consumer EPR [" +
                                                       neighborBrokerConsumerEprUrl + "]");
                        }
                        BrokerConsumerReference bcr = new BrokerConsumerReference();
                        bcr.setReference(EncodingUtils.createEndpointReference(neighborBrokerStruct.consumerUrl, neighborBrokerStruct.brokerID));
                        BrokerProducerReference bpr = new BrokerProducerReference();
                        bpr.setReference(EncodingUtils.createEndpointReference(neighborBrokerStruct.producerUrl, neighborBrokerStruct.brokerID));
                        Broker neighborBroker = new Broker();
                        neighborBroker.setBrokerConsumerReference(bcr);
                        neighborBroker.setBrokerProducerReference(bpr);
                        neighborBrokers.add(neighborBroker);
                    }
                    int lastIndex = neighborBrokers.size() - 1;
                    for (int index = lastIndex; index >= 0; index--) {
                        List<Broker> writeableNeighborBrokers = ((index > 0) ? neighborBrokers.subList(0, index) : null);
                        WriteableReplaceBrokerConnection wrbc = new WriteableReplaceBrokerConnection(rbEpr, writeableNeighborBrokers);
                        URL targetUrl =
                            neighborBrokers.get(index).getBrokerProducerReference().getReference().getEndpointAddress().getAddress();
                        try {
                            IOUtils.sendHttpRequest(targetUrl, Constants.REPLACE_BROKER_CONNECTION_OP, wrbc, null);
                        } catch(Exception e) {
                            e.printStackTrace();
                            throw new RuntimeException(e);
                        }
                    }
                }
                
                BrokerStruct removedBrokerStruct = null;
                URL rbEprUrl = rbEpr.getEndpointAddress().getAddress();
                for (BrokerStruct brokerSruct : brokerList) {
                    if (brokerSruct.consumerUrl.equals(rbEprUrl)) {
                        removedBrokerStruct = brokerSruct;
                        break;
                    }
                }
                if (removedBrokerStruct == null) {
                    throw new RuntimeException("Can't find broker to remove for EPR [" + rbEprUrl + "]");
                }
                if(!brokerList.remove(removedBrokerStruct)) {
                    throw new RuntimeException("Broker was not removed");
                }
            }
        }
    }
    
    class NotificationTypeLock {
        public boolean isLocked;
    }
    
    class WriteableEPW implements Writeable {
        private EndpointReferenceWrapper epw;
        
        public WriteableEPW(EndpointReferenceWrapper epw, URL url) {
            EndpointAddress epa = new EndpointAddress();
            epa.setAddress(url);
            EndpointReference epr = new EndpointReference();
            epr.setEndpointAddress(epa);
            epw.setReference(epr);
            this.epw = epw;
        }
        
        public void write(OutputStream os) throws IOUtilsException {
            EncodingUtils.encodeToStream(encodingRegistry, epw, os);
        }
    }
    
    class InputStreamDecoder implements ReadableContinuation {
        
        public Object read(InputStream istream) throws IOUtilsException {
            try {
                return EncodingUtils.decodeFromStream(encodingRegistry, istream);
            } catch(EncodingException e) {
                throw new IOUtilsException(e);
            }
        }
    }
    
    class BrokerStruct {
        public URL consumerUrl;
        public URL producerUrl;
        public String brokerID;
        
        public BrokerStruct(URL consumerUrl, URL producerUrl, String brokerID) {
            this.consumerUrl = consumerUrl;
            this.producerUrl = producerUrl;
            this.brokerID = brokerID;
        }
    }
    
    class WriteableNewBroker implements Writeable {
        private NewBroker newBroker;
        
        public WriteableNewBroker(URL consumerUrl, URL producerUrl, String brokerID) {
            newBroker = new NewBroker();
            BrokerConsumerReference bcr = new BrokerConsumerReference();
            bcr.setReference(EncodingUtils.createEndpointReference(consumerUrl, brokerID));
            newBroker.setBrokerConsumerReference(bcr);
            
            BrokerProducerReference bpr = new BrokerProducerReference();
            bpr.setReference(EncodingUtils.createEndpointReference(producerUrl, brokerID));
            newBroker.setBrokerProducerReference(bpr);
        }
        
        public void write(OutputStream os) throws IOUtilsException {
            EncodingUtils.encodeToStream(encodingRegistry, newBroker, os);
        }
    }

    class WriteableNewBrokerAck implements Writeable {
        public void write(OutputStream os) throws IOUtilsException {
            EncodingUtils.encodeToStream(encodingRegistry, new NewBrokerAck(), os);
        }
    }
    
    class WriteableRemoveBroker implements Writeable {
        private RemoveBroker removeBroker;
        
        public WriteableRemoveBroker(EndpointReference brokerConsumerEpr, List<EndpointReference> neighborBrokerConsumerEprs) {
            removeBroker = new RemoveBroker();
            BrokerConsumerReference brokerConsumerReference = new BrokerConsumerReference();
            brokerConsumerReference.setReference(brokerConsumerEpr);
            removeBroker.setBrokerConsumerReference(brokerConsumerReference);
            if (neighborBrokerConsumerEprs != null) {
                NeighborBrokerConsumers neighborBrokerConsumers = new NeighborBrokerConsumers();
                neighborBrokerConsumers.setReferenceSequence(neighborBrokerConsumerEprs);
                neighborBrokerConsumers.setSequenceType(Constants.BrokerConsumers);
                removeBroker.setNeighborBrokerConsumers(neighborBrokerConsumers);
            }
        }
        
        public void write(OutputStream os) throws IOUtilsException {
            EncodingUtils.encodeToStream(encodingRegistry, removeBroker, os);
        }
    }
    
    class WriteableReplaceBrokerConnection implements Writeable {
        private ReplaceBrokerConnection replaceBrokerConnection;
        
        public WriteableReplaceBrokerConnection(EndpointReference removedBrokerEpr, List<Broker> brokerSequence) {
            replaceBrokerConnection = new ReplaceBrokerConnection();
            RemovedBroker removedBroker = new RemovedBroker();
            removedBroker.setReference(removedBrokerEpr);
            replaceBrokerConnection.setRemovedBroker(removedBroker);
            if (brokerSequence != null) {
                Neighbors neighbors = new Neighbors();
                neighbors.setBrokerSequence(brokerSequence);
                replaceBrokerConnection.setNeighbors(neighbors);
            }
        }
        
        public void write(OutputStream os) throws IOUtilsException {
            EncodingUtils.encodeToStream(encodingRegistry, replaceBrokerConnection, os);
        }
    }
}