001/**
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements.  See the NOTICE file distributed with
004 * this work for additional information regarding copyright ownership.
005 * The ASF licenses this file to You under the Apache License, Version 2.0
006 * (the "License"); you may not use this file except in compliance with
007 * the License.  You may obtain a copy of the License at
008 *
009 *      http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the License for the specific language governing permissions and
015 * limitations under the License.
016 */
017package org.apache.activemq.broker;
018
019import java.io.EOFException;
020import java.io.IOException;
021import java.net.SocketException;
022import java.net.URI;
023import java.util.Collection;
024import java.util.HashMap;
025import java.util.Iterator;
026import java.util.LinkedList;
027import java.util.List;
028import java.util.Map;
029import java.util.Properties;
030import java.util.concurrent.ConcurrentHashMap;
031import java.util.concurrent.CopyOnWriteArrayList;
032import java.util.concurrent.CountDownLatch;
033import java.util.concurrent.TimeUnit;
034import java.util.concurrent.atomic.AtomicBoolean;
035import java.util.concurrent.atomic.AtomicInteger;
036import java.util.concurrent.atomic.AtomicReference;
037import java.util.concurrent.locks.ReentrantReadWriteLock;
038
039import javax.transaction.xa.XAResource;
040
041import org.apache.activemq.advisory.AdvisorySupport;
042import org.apache.activemq.broker.region.ConnectionStatistics;
043import org.apache.activemq.broker.region.RegionBroker;
044import org.apache.activemq.command.ActiveMQDestination;
045import org.apache.activemq.command.BrokerInfo;
046import org.apache.activemq.command.Command;
047import org.apache.activemq.command.CommandTypes;
048import org.apache.activemq.command.ConnectionControl;
049import org.apache.activemq.command.ConnectionError;
050import org.apache.activemq.command.ConnectionId;
051import org.apache.activemq.command.ConnectionInfo;
052import org.apache.activemq.command.ConsumerControl;
053import org.apache.activemq.command.ConsumerId;
054import org.apache.activemq.command.ConsumerInfo;
055import org.apache.activemq.command.ControlCommand;
056import org.apache.activemq.command.DataArrayResponse;
057import org.apache.activemq.command.DestinationInfo;
058import org.apache.activemq.command.ExceptionResponse;
059import org.apache.activemq.command.FlushCommand;
060import org.apache.activemq.command.IntegerResponse;
061import org.apache.activemq.command.KeepAliveInfo;
062import org.apache.activemq.command.Message;
063import org.apache.activemq.command.MessageAck;
064import org.apache.activemq.command.MessageDispatch;
065import org.apache.activemq.command.MessageDispatchNotification;
066import org.apache.activemq.command.MessagePull;
067import org.apache.activemq.command.ProducerAck;
068import org.apache.activemq.command.ProducerId;
069import org.apache.activemq.command.ProducerInfo;
070import org.apache.activemq.command.RemoveInfo;
071import org.apache.activemq.command.RemoveSubscriptionInfo;
072import org.apache.activemq.command.Response;
073import org.apache.activemq.command.SessionId;
074import org.apache.activemq.command.SessionInfo;
075import org.apache.activemq.command.ShutdownInfo;
076import org.apache.activemq.command.TransactionId;
077import org.apache.activemq.command.TransactionInfo;
078import org.apache.activemq.command.WireFormatInfo;
079import org.apache.activemq.network.DemandForwardingBridge;
080import org.apache.activemq.network.MBeanNetworkListener;
081import org.apache.activemq.network.NetworkBridgeConfiguration;
082import org.apache.activemq.network.NetworkBridgeFactory;
083import org.apache.activemq.security.MessageAuthorizationPolicy;
084import org.apache.activemq.state.CommandVisitor;
085import org.apache.activemq.state.ConnectionState;
086import org.apache.activemq.state.ConsumerState;
087import org.apache.activemq.state.ProducerState;
088import org.apache.activemq.state.SessionState;
089import org.apache.activemq.state.TransactionState;
090import org.apache.activemq.thread.Task;
091import org.apache.activemq.thread.TaskRunner;
092import org.apache.activemq.thread.TaskRunnerFactory;
093import org.apache.activemq.transaction.Transaction;
094import org.apache.activemq.transport.DefaultTransportListener;
095import org.apache.activemq.transport.ResponseCorrelator;
096import org.apache.activemq.transport.TransmitCallback;
097import org.apache.activemq.transport.Transport;
098import org.apache.activemq.transport.TransportDisposedIOException;
099import org.apache.activemq.util.IntrospectionSupport;
100import org.apache.activemq.util.MarshallingSupport;
101import org.slf4j.Logger;
102import org.slf4j.LoggerFactory;
103import org.slf4j.MDC;
104
105public class TransportConnection implements Connection, Task, CommandVisitor {
106    private static final Logger LOG = LoggerFactory.getLogger(TransportConnection.class);
107    private static final Logger TRANSPORTLOG = LoggerFactory.getLogger(TransportConnection.class.getName() + ".Transport");
108    private static final Logger SERVICELOG = LoggerFactory.getLogger(TransportConnection.class.getName() + ".Service");
109    // Keeps track of the broker and connector that created this connection.
110    protected final Broker broker;
111    protected final BrokerService brokerService;
112    protected final TransportConnector connector;
113    // Keeps track of the state of the connections.
114    // protected final ConcurrentHashMap localConnectionStates=new
115    // ConcurrentHashMap();
116    protected final Map<ConnectionId, ConnectionState> brokerConnectionStates;
117    // The broker and wireformat info that was exchanged.
118    protected BrokerInfo brokerInfo;
119    protected final List<Command> dispatchQueue = new LinkedList<Command>();
120    protected TaskRunner taskRunner;
121    protected final AtomicReference<Throwable> transportException = new AtomicReference<Throwable>();
122    protected AtomicBoolean dispatchStopped = new AtomicBoolean(false);
123    private final Transport transport;
124    private MessageAuthorizationPolicy messageAuthorizationPolicy;
125    private WireFormatInfo wireFormatInfo;
126    // Used to do async dispatch.. this should perhaps be pushed down into the
127    // transport layer..
128    private boolean inServiceException;
129    private final ConnectionStatistics statistics = new ConnectionStatistics();
130    private boolean manageable;
131    private boolean slow;
132    private boolean markedCandidate;
133    private boolean blockedCandidate;
134    private boolean blocked;
135    private boolean connected;
136    private boolean active;
137
138    // state management around pending stop
139    private static final int NEW           = 0;
140    private static final int STARTING      = 1;
141    private static final int STARTED       = 2;
142    private static final int PENDING_STOP  = 3;
143    private final AtomicInteger status = new AtomicInteger(NEW);
144
145    private long timeStamp;
146    private final AtomicBoolean stopping = new AtomicBoolean(false);
147    private final CountDownLatch stopped = new CountDownLatch(1);
148    private final AtomicBoolean asyncException = new AtomicBoolean(false);
149    private final Map<ProducerId, ProducerBrokerExchange> producerExchanges = new HashMap<ProducerId, ProducerBrokerExchange>();
150    private final Map<ConsumerId, ConsumerBrokerExchange> consumerExchanges = new HashMap<ConsumerId, ConsumerBrokerExchange>();
151    private final CountDownLatch dispatchStoppedLatch = new CountDownLatch(1);
152    private ConnectionContext context;
153    private boolean networkConnection;
154    private boolean faultTolerantConnection;
155    private final AtomicInteger protocolVersion = new AtomicInteger(CommandTypes.PROTOCOL_VERSION);
156    private DemandForwardingBridge duplexBridge;
157    private final TaskRunnerFactory taskRunnerFactory;
158    private final TaskRunnerFactory stopTaskRunnerFactory;
159    private TransportConnectionStateRegister connectionStateRegister = new SingleTransportConnectionStateRegister();
160    private final ReentrantReadWriteLock serviceLock = new ReentrantReadWriteLock();
161    private String duplexNetworkConnectorId;
162
163    /**
164     * @param taskRunnerFactory - can be null if you want direct dispatch to the transport
165     *                          else commands are sent async.
166     * @param stopTaskRunnerFactory - can <b>not</b> be null, used for stopping this connection.
167     */
168    public TransportConnection(TransportConnector connector, final Transport transport, Broker broker,
169                               TaskRunnerFactory taskRunnerFactory, TaskRunnerFactory stopTaskRunnerFactory) {
170        this.connector = connector;
171        this.broker = broker;
172        this.brokerService = broker.getBrokerService();
173
174        RegionBroker rb = (RegionBroker) broker.getAdaptor(RegionBroker.class);
175        brokerConnectionStates = rb.getConnectionStates();
176        if (connector != null) {
177            this.statistics.setParent(connector.getStatistics());
178            this.messageAuthorizationPolicy = connector.getMessageAuthorizationPolicy();
179        }
180        this.taskRunnerFactory = taskRunnerFactory;
181        this.stopTaskRunnerFactory = stopTaskRunnerFactory;
182        this.transport = transport;
183        if( this.transport instanceof BrokerServiceAware ) {
184            ((BrokerServiceAware)this.transport).setBrokerService(brokerService);
185        }
186        this.transport.setTransportListener(new DefaultTransportListener() {
187            @Override
188            public void onCommand(Object o) {
189                serviceLock.readLock().lock();
190                try {
191                    if (!(o instanceof Command)) {
192                        throw new RuntimeException("Protocol violation - Command corrupted: " + o.toString());
193                    }
194                    Command command = (Command) o;
195                    if (!brokerService.isStopping()) {
196                        Response response = service(command);
197                        if (response != null && !brokerService.isStopping()) {
198                            dispatchSync(response);
199                        }
200                    } else {
201                        throw new BrokerStoppedException("Broker " + brokerService + " is being stopped");
202                    }
203                } finally {
204                    serviceLock.readLock().unlock();
205                }
206            }
207
208            @Override
209            public void onException(IOException exception) {
210                serviceLock.readLock().lock();
211                try {
212                    serviceTransportException(exception);
213                } finally {
214                    serviceLock.readLock().unlock();
215                }
216            }
217        });
218        connected = true;
219    }
220
221    /**
222     * Returns the number of messages to be dispatched to this connection
223     *
224     * @return size of dispatch queue
225     */
226    @Override
227    public int getDispatchQueueSize() {
228        synchronized (dispatchQueue) {
229            return dispatchQueue.size();
230        }
231    }
232
233    public void serviceTransportException(IOException e) {
234        if (!stopping.get() && status.get() != PENDING_STOP) {
235            transportException.set(e);
236            if (TRANSPORTLOG.isDebugEnabled()) {
237                TRANSPORTLOG.debug(this + " failed: " + e, e);
238            } else if (TRANSPORTLOG.isWarnEnabled() && !suppressed(e)) {
239                TRANSPORTLOG.warn(this + " failed: " + e);
240            }
241            stopAsync(e);
242        }
243    }
244
245    private boolean suppressed(IOException e) {
246        return !connector.isWarnOnRemoteClose() && ((e instanceof SocketException && e.getMessage().indexOf("reset") != -1) || e instanceof EOFException);
247    }
248
249    /**
250     * Calls the serviceException method in an async thread. Since handling a
251     * service exception closes a socket, we should not tie up broker threads
252     * since client sockets may hang or cause deadlocks.
253     */
254    @Override
255    public void serviceExceptionAsync(final IOException e) {
256        if (asyncException.compareAndSet(false, true)) {
257            new Thread("Async Exception Handler") {
258                @Override
259                public void run() {
260                    serviceException(e);
261                }
262            }.start();
263        }
264    }
265
266    /**
267     * Closes a clients connection due to a detected error. Errors are ignored
268     * if: the client is closing or broker is closing. Otherwise, the connection
269     * error transmitted to the client before stopping it's transport.
270     */
271    @Override
272    public void serviceException(Throwable e) {
273        // are we a transport exception such as not being able to dispatch
274        // synchronously to a transport
275        if (e instanceof IOException) {
276            serviceTransportException((IOException) e);
277        } else if (e.getClass() == BrokerStoppedException.class) {
278            // Handle the case where the broker is stopped
279            // But the client is still connected.
280            if (!stopping.get()) {
281                SERVICELOG.debug("Broker has been stopped.  Notifying client and closing his connection.");
282                ConnectionError ce = new ConnectionError();
283                ce.setException(e);
284                dispatchSync(ce);
285                // Record the error that caused the transport to stop
286                transportException.set(e);
287                // Wait a little bit to try to get the output buffer to flush
288                // the exception notification to the client.
289                try {
290                    Thread.sleep(500);
291                } catch (InterruptedException ie) {
292                    Thread.currentThread().interrupt();
293                }
294                // Worst case is we just kill the connection before the
295                // notification gets to him.
296                stopAsync();
297            }
298        } else if (!stopping.get() && !inServiceException) {
299            inServiceException = true;
300            try {
301                if (SERVICELOG.isDebugEnabled()) {
302                    SERVICELOG.debug("Async error occurred: " + e, e);
303                } else {
304                    SERVICELOG.warn("Async error occurred: " + e);
305                }
306                ConnectionError ce = new ConnectionError();
307                ce.setException(e);
308                if (status.get() == PENDING_STOP) {
309                    dispatchSync(ce);
310                } else {
311                    dispatchAsync(ce);
312                }
313            } finally {
314                inServiceException = false;
315            }
316        }
317    }
318
319    @Override
320    public Response service(Command command) {
321        MDC.put("activemq.connector", connector.getUri().toString());
322        Response response = null;
323        boolean responseRequired = command.isResponseRequired();
324        int commandId = command.getCommandId();
325        try {
326            if (status.get() != PENDING_STOP) {
327                response = command.visit(this);
328            } else {
329                response = new ExceptionResponse(transportException.get());
330            }
331        } catch (Throwable e) {
332            if (SERVICELOG.isDebugEnabled() && e.getClass() != BrokerStoppedException.class) {
333                SERVICELOG.debug("Error occured while processing " + (responseRequired ? "sync" : "async")
334                        + " command: " + command + ", exception: " + e, e);
335            }
336
337            if (e instanceof SuppressReplyException || (e.getCause() instanceof SuppressReplyException)) {
338                LOG.info("Suppressing reply to: " + command + " on: " + e + ", cause: " + e.getCause());
339                responseRequired = false;
340            }
341
342            if (responseRequired) {
343                if (e instanceof SecurityException || e.getCause() instanceof SecurityException) {
344                    SERVICELOG.warn("Security Error occurred on connection to: {}, {}",
345                            transport.getRemoteAddress(), e.getMessage());
346                }
347                response = new ExceptionResponse(e);
348            } else {
349                forceRollbackOnlyOnFailedAsyncTransactionOp(e, command);
350                serviceException(e);
351            }
352        }
353        if (responseRequired) {
354            if (response == null) {
355                response = new Response();
356            }
357            response.setCorrelationId(commandId);
358        }
359        // The context may have been flagged so that the response is not
360        // sent.
361        if (context != null) {
362            if (context.isDontSendReponse()) {
363                context.setDontSendReponse(false);
364                response = null;
365            }
366            context = null;
367        }
368        MDC.remove("activemq.connector");
369        return response;
370    }
371
372    private void forceRollbackOnlyOnFailedAsyncTransactionOp(Throwable e, Command command) {
373        if (brokerService.isRollbackOnlyOnAsyncException() && !(e instanceof IOException) && isInTransaction(command)) {
374            Transaction transaction = getActiveTransaction(command);
375            if (transaction != null && !transaction.isRollbackOnly()) {
376                LOG.debug("on async exception, force rollback of transaction for: " + command, e);
377                transaction.setRollbackOnly(e);
378            }
379        }
380    }
381
382    private Transaction getActiveTransaction(Command command) {
383        Transaction transaction = null;
384        try {
385            if (command instanceof Message) {
386                Message messageSend = (Message) command;
387                ProducerId producerId = messageSend.getProducerId();
388                ProducerBrokerExchange producerExchange = getProducerBrokerExchange(producerId);
389                transaction = producerExchange.getConnectionContext().getTransactions().get(messageSend.getTransactionId());
390            } else if (command instanceof  MessageAck) {
391                MessageAck messageAck = (MessageAck) command;
392                ConsumerBrokerExchange consumerExchange = getConsumerBrokerExchange(messageAck.getConsumerId());
393                if (consumerExchange != null) {
394                    transaction = consumerExchange.getConnectionContext().getTransactions().get(messageAck.getTransactionId());
395                }
396            }
397        } catch(Exception ignored){
398            LOG.trace("failed to find active transaction for command: " + command, ignored);
399        }
400        return transaction;
401    }
402
403    private boolean isInTransaction(Command command) {
404        return command instanceof Message && ((Message)command).isInTransaction()
405                || command instanceof MessageAck && ((MessageAck)command).isInTransaction();
406    }
407
408    @Override
409    public Response processKeepAlive(KeepAliveInfo info) throws Exception {
410        return null;
411    }
412
413    @Override
414    public Response processRemoveSubscription(RemoveSubscriptionInfo info) throws Exception {
415        broker.removeSubscription(lookupConnectionState(info.getConnectionId()).getContext(), info);
416        return null;
417    }
418
419    @Override
420    public Response processWireFormat(WireFormatInfo info) throws Exception {
421        wireFormatInfo = info;
422        protocolVersion.set(info.getVersion());
423        return null;
424    }
425
426    @Override
427    public Response processShutdown(ShutdownInfo info) throws Exception {
428        stopAsync();
429        return null;
430    }
431
432    @Override
433    public Response processFlush(FlushCommand command) throws Exception {
434        return null;
435    }
436
437    @Override
438    public Response processBeginTransaction(TransactionInfo info) throws Exception {
439        TransportConnectionState cs = lookupConnectionState(info.getConnectionId());
440        context = null;
441        if (cs != null) {
442            context = cs.getContext();
443        }
444        if (cs == null) {
445            throw new NullPointerException("Context is null");
446        }
447        // Avoid replaying dup commands
448        if (cs.getTransactionState(info.getTransactionId()) == null) {
449            cs.addTransactionState(info.getTransactionId());
450            broker.beginTransaction(context, info.getTransactionId());
451        }
452        return null;
453    }
454
455    @Override
456    public int getActiveTransactionCount() {
457        int rc = 0;
458        for (TransportConnectionState cs : connectionStateRegister.listConnectionStates()) {
459            Collection<TransactionState> transactions = cs.getTransactionStates();
460            for (TransactionState transaction : transactions) {
461                rc++;
462            }
463        }
464        return rc;
465    }
466
467    @Override
468    public Long getOldestActiveTransactionDuration() {
469        TransactionState oldestTX = null;
470        for (TransportConnectionState cs : connectionStateRegister.listConnectionStates()) {
471            Collection<TransactionState> transactions = cs.getTransactionStates();
472            for (TransactionState transaction : transactions) {
473                if( oldestTX ==null || oldestTX.getCreatedAt() < transaction.getCreatedAt() ) {
474                    oldestTX = transaction;
475                }
476            }
477        }
478        if( oldestTX == null ) {
479            return null;
480        }
481        return System.currentTimeMillis() - oldestTX.getCreatedAt();
482    }
483
484    @Override
485    public Response processEndTransaction(TransactionInfo info) throws Exception {
486        // No need to do anything. This packet is just sent by the client
487        // make sure he is synced with the server as commit command could
488        // come from a different connection.
489        return null;
490    }
491
492    @Override
493    public Response processPrepareTransaction(TransactionInfo info) throws Exception {
494        TransportConnectionState cs = lookupConnectionState(info.getConnectionId());
495        context = null;
496        if (cs != null) {
497            context = cs.getContext();
498        }
499        if (cs == null) {
500            throw new NullPointerException("Context is null");
501        }
502        TransactionState transactionState = cs.getTransactionState(info.getTransactionId());
503        if (transactionState == null) {
504            throw new IllegalStateException("Cannot prepare a transaction that had not been started or previously returned XA_RDONLY: "
505                    + info.getTransactionId());
506        }
507        // Avoid dups.
508        if (!transactionState.isPrepared()) {
509            transactionState.setPrepared(true);
510            int result = broker.prepareTransaction(context, info.getTransactionId());
511            transactionState.setPreparedResult(result);
512            if (result == XAResource.XA_RDONLY) {
513                // we are done, no further rollback or commit from TM
514                cs.removeTransactionState(info.getTransactionId());
515            }
516            IntegerResponse response = new IntegerResponse(result);
517            return response;
518        } else {
519            IntegerResponse response = new IntegerResponse(transactionState.getPreparedResult());
520            return response;
521        }
522    }
523
524    @Override
525    public Response processCommitTransactionOnePhase(TransactionInfo info) throws Exception {
526        TransportConnectionState cs = lookupConnectionState(info.getConnectionId());
527        context = cs.getContext();
528        cs.removeTransactionState(info.getTransactionId());
529        broker.commitTransaction(context, info.getTransactionId(), true);
530        return null;
531    }
532
533    @Override
534    public Response processCommitTransactionTwoPhase(TransactionInfo info) throws Exception {
535        TransportConnectionState cs = lookupConnectionState(info.getConnectionId());
536        context = cs.getContext();
537        cs.removeTransactionState(info.getTransactionId());
538        broker.commitTransaction(context, info.getTransactionId(), false);
539        return null;
540    }
541
542    @Override
543    public Response processRollbackTransaction(TransactionInfo info) throws Exception {
544        TransportConnectionState cs = lookupConnectionState(info.getConnectionId());
545        context = cs.getContext();
546        cs.removeTransactionState(info.getTransactionId());
547        broker.rollbackTransaction(context, info.getTransactionId());
548        return null;
549    }
550
551    @Override
552    public Response processForgetTransaction(TransactionInfo info) throws Exception {
553        TransportConnectionState cs = lookupConnectionState(info.getConnectionId());
554        context = cs.getContext();
555        broker.forgetTransaction(context, info.getTransactionId());
556        return null;
557    }
558
559    @Override
560    public Response processRecoverTransactions(TransactionInfo info) throws Exception {
561        TransportConnectionState cs = lookupConnectionState(info.getConnectionId());
562        context = cs.getContext();
563        TransactionId[] preparedTransactions = broker.getPreparedTransactions(context);
564        return new DataArrayResponse(preparedTransactions);
565    }
566
567    @Override
568    public Response processMessage(Message messageSend) throws Exception {
569        ProducerId producerId = messageSend.getProducerId();
570        ProducerBrokerExchange producerExchange = getProducerBrokerExchange(producerId);
571        if (producerExchange.canDispatch(messageSend)) {
572            broker.send(producerExchange, messageSend);
573        }
574        return null;
575    }
576
577    @Override
578    public Response processMessageAck(MessageAck ack) throws Exception {
579        ConsumerBrokerExchange consumerExchange = getConsumerBrokerExchange(ack.getConsumerId());
580        if (consumerExchange != null) {
581            broker.acknowledge(consumerExchange, ack);
582        } else if (ack.isInTransaction()) {
583            LOG.warn("no matching consumer, ignoring ack {}", consumerExchange, ack);
584        }
585        return null;
586    }
587
588    @Override
589    public Response processMessagePull(MessagePull pull) throws Exception {
590        return broker.messagePull(lookupConnectionState(pull.getConsumerId()).getContext(), pull);
591    }
592
593    @Override
594    public Response processMessageDispatchNotification(MessageDispatchNotification notification) throws Exception {
595        broker.processDispatchNotification(notification);
596        return null;
597    }
598
599    @Override
600    public Response processAddDestination(DestinationInfo info) throws Exception {
601        TransportConnectionState cs = lookupConnectionState(info.getConnectionId());
602        broker.addDestinationInfo(cs.getContext(), info);
603        if (info.getDestination().isTemporary()) {
604            cs.addTempDestination(info);
605        }
606        return null;
607    }
608
609    @Override
610    public Response processRemoveDestination(DestinationInfo info) throws Exception {
611        TransportConnectionState cs = lookupConnectionState(info.getConnectionId());
612        broker.removeDestinationInfo(cs.getContext(), info);
613        if (info.getDestination().isTemporary()) {
614            cs.removeTempDestination(info.getDestination());
615        }
616        return null;
617    }
618
619    @Override
620    public Response processAddProducer(ProducerInfo info) throws Exception {
621        SessionId sessionId = info.getProducerId().getParentId();
622        ConnectionId connectionId = sessionId.getParentId();
623        TransportConnectionState cs = lookupConnectionState(connectionId);
624        if (cs == null) {
625            throw new IllegalStateException("Cannot add a producer to a connection that had not been registered: "
626                    + connectionId);
627        }
628        SessionState ss = cs.getSessionState(sessionId);
629        if (ss == null) {
630            throw new IllegalStateException("Cannot add a producer to a session that had not been registered: "
631                    + sessionId);
632        }
633        // Avoid replaying dup commands
634        if (!ss.getProducerIds().contains(info.getProducerId())) {
635            ActiveMQDestination destination = info.getDestination();
636            // Do not check for null here as it would cause the count of max producers to exclude
637            // anonymous producers.  The isAdvisoryTopic method checks for null so it is safe to
638            // call it from here with a null Destination value.
639            if (!AdvisorySupport.isAdvisoryTopic(destination)) {
640                if (getProducerCount(connectionId) >= connector.getMaximumProducersAllowedPerConnection()){
641                    throw new IllegalStateException("Can't add producer on connection " + connectionId + ": at maximum limit: " + connector.getMaximumProducersAllowedPerConnection());
642                }
643            }
644            broker.addProducer(cs.getContext(), info);
645            try {
646                ss.addProducer(info);
647            } catch (IllegalStateException e) {
648                broker.removeProducer(cs.getContext(), info);
649            }
650
651        }
652        return null;
653    }
654
655    @Override
656    public Response processRemoveProducer(ProducerId id) throws Exception {
657        SessionId sessionId = id.getParentId();
658        ConnectionId connectionId = sessionId.getParentId();
659        TransportConnectionState cs = lookupConnectionState(connectionId);
660        SessionState ss = cs.getSessionState(sessionId);
661        if (ss == null) {
662            throw new IllegalStateException("Cannot remove a producer from a session that had not been registered: "
663                    + sessionId);
664        }
665        ProducerState ps = ss.removeProducer(id);
666        if (ps == null) {
667            throw new IllegalStateException("Cannot remove a producer that had not been registered: " + id);
668        }
669        removeProducerBrokerExchange(id);
670        broker.removeProducer(cs.getContext(), ps.getInfo());
671        return null;
672    }
673
674    @Override
675    public Response processAddConsumer(ConsumerInfo info) throws Exception {
676        SessionId sessionId = info.getConsumerId().getParentId();
677        ConnectionId connectionId = sessionId.getParentId();
678        TransportConnectionState cs = lookupConnectionState(connectionId);
679        if (cs == null) {
680            throw new IllegalStateException("Cannot add a consumer to a connection that had not been registered: "
681                    + connectionId);
682        }
683        SessionState ss = cs.getSessionState(sessionId);
684        if (ss == null) {
685            throw new IllegalStateException(broker.getBrokerName()
686                    + " Cannot add a consumer to a session that had not been registered: " + sessionId);
687        }
688        // Avoid replaying dup commands
689        if (!ss.getConsumerIds().contains(info.getConsumerId())) {
690            ActiveMQDestination destination = info.getDestination();
691            if (destination != null && !AdvisorySupport.isAdvisoryTopic(destination)) {
692                if (getConsumerCount(connectionId) >= connector.getMaximumConsumersAllowedPerConnection()){
693                    throw new IllegalStateException("Can't add consumer on connection " + connectionId + ": at maximum limit: " + connector.getMaximumConsumersAllowedPerConnection());
694                }
695            }
696
697            broker.addConsumer(cs.getContext(), info);
698            try {
699                ss.addConsumer(info);
700                addConsumerBrokerExchange(cs, info.getConsumerId());
701            } catch (IllegalStateException e) {
702                broker.removeConsumer(cs.getContext(), info);
703            }
704
705        }
706        return null;
707    }
708
709    @Override
710    public Response processRemoveConsumer(ConsumerId id, long lastDeliveredSequenceId) throws Exception {
711        SessionId sessionId = id.getParentId();
712        ConnectionId connectionId = sessionId.getParentId();
713        TransportConnectionState cs = lookupConnectionState(connectionId);
714        if (cs == null) {
715            throw new IllegalStateException("Cannot remove a consumer from a connection that had not been registered: "
716                    + connectionId);
717        }
718        SessionState ss = cs.getSessionState(sessionId);
719        if (ss == null) {
720            throw new IllegalStateException("Cannot remove a consumer from a session that had not been registered: "
721                    + sessionId);
722        }
723        ConsumerState consumerState = ss.removeConsumer(id);
724        if (consumerState == null) {
725            throw new IllegalStateException("Cannot remove a consumer that had not been registered: " + id);
726        }
727        ConsumerInfo info = consumerState.getInfo();
728        info.setLastDeliveredSequenceId(lastDeliveredSequenceId);
729        broker.removeConsumer(cs.getContext(), consumerState.getInfo());
730        removeConsumerBrokerExchange(id);
731        return null;
732    }
733
734    @Override
735    public Response processAddSession(SessionInfo info) throws Exception {
736        ConnectionId connectionId = info.getSessionId().getParentId();
737        TransportConnectionState cs = lookupConnectionState(connectionId);
738        // Avoid replaying dup commands
739        if (cs != null && !cs.getSessionIds().contains(info.getSessionId())) {
740            broker.addSession(cs.getContext(), info);
741            try {
742                cs.addSession(info);
743            } catch (IllegalStateException e) {
744                LOG.warn("Failed to add session: {}", info.getSessionId(), e);
745                broker.removeSession(cs.getContext(), info);
746            }
747        }
748        return null;
749    }
750
751    @Override
752    public Response processRemoveSession(SessionId id, long lastDeliveredSequenceId) throws Exception {
753        ConnectionId connectionId = id.getParentId();
754        TransportConnectionState cs = lookupConnectionState(connectionId);
755        if (cs == null) {
756            throw new IllegalStateException("Cannot remove session from connection that had not been registered: " + connectionId);
757        }
758        SessionState session = cs.getSessionState(id);
759        if (session == null) {
760            throw new IllegalStateException("Cannot remove session that had not been registered: " + id);
761        }
762        // Don't let new consumers or producers get added while we are closing
763        // this down.
764        session.shutdown();
765        // Cascade the connection stop to the consumers and producers.
766        for (ConsumerId consumerId : session.getConsumerIds()) {
767            try {
768                processRemoveConsumer(consumerId, lastDeliveredSequenceId);
769            } catch (Throwable e) {
770                LOG.warn("Failed to remove consumer: {}", consumerId, e);
771            }
772        }
773        for (ProducerId producerId : session.getProducerIds()) {
774            try {
775                processRemoveProducer(producerId);
776            } catch (Throwable e) {
777                LOG.warn("Failed to remove producer: {}", producerId, e);
778            }
779        }
780        cs.removeSession(id);
781        broker.removeSession(cs.getContext(), session.getInfo());
782        return null;
783    }
784
785    @Override
786    public Response processAddConnection(ConnectionInfo info) throws Exception {
787        // Older clients should have been defaulting this field to true.. but
788        // they were not.
789        if (wireFormatInfo != null && wireFormatInfo.getVersion() <= 2) {
790            info.setClientMaster(true);
791        }
792        TransportConnectionState state;
793        // Make sure 2 concurrent connections by the same ID only generate 1
794        // TransportConnectionState object.
795        synchronized (brokerConnectionStates) {
796            state = (TransportConnectionState) brokerConnectionStates.get(info.getConnectionId());
797            if (state == null) {
798                state = new TransportConnectionState(info, this);
799                brokerConnectionStates.put(info.getConnectionId(), state);
800            }
801            state.incrementReference();
802        }
803        // If there are 2 concurrent connections for the same connection id,
804        // then last one in wins, we need to sync here
805        // to figure out the winner.
806        synchronized (state.getConnectionMutex()) {
807            if (state.getConnection() != this) {
808                LOG.debug("Killing previous stale connection: {}", state.getConnection().getRemoteAddress());
809                state.getConnection().stop();
810                LOG.debug("Connection {} taking over previous connection: {}", getRemoteAddress(), state.getConnection().getRemoteAddress());
811                state.setConnection(this);
812                state.reset(info);
813            }
814        }
815        registerConnectionState(info.getConnectionId(), state);
816        LOG.debug("Setting up new connection id: {}, address: {}, info: {}", new Object[]{ info.getConnectionId(), getRemoteAddress(), info });
817        this.faultTolerantConnection = info.isFaultTolerant();
818        // Setup the context.
819        String clientId = info.getClientId();
820        context = new ConnectionContext();
821        context.setBroker(broker);
822        context.setClientId(clientId);
823        context.setClientMaster(info.isClientMaster());
824        context.setConnection(this);
825        context.setConnectionId(info.getConnectionId());
826        context.setConnector(connector);
827        context.setMessageAuthorizationPolicy(getMessageAuthorizationPolicy());
828        context.setNetworkConnection(networkConnection);
829        context.setFaultTolerant(faultTolerantConnection);
830        context.setTransactions(new ConcurrentHashMap<TransactionId, Transaction>());
831        context.setUserName(info.getUserName());
832        context.setWireFormatInfo(wireFormatInfo);
833        context.setReconnect(info.isFailoverReconnect());
834        this.manageable = info.isManageable();
835        context.setConnectionState(state);
836        state.setContext(context);
837        state.setConnection(this);
838        if (info.getClientIp() == null) {
839            info.setClientIp(getRemoteAddress());
840        }
841
842        try {
843            broker.addConnection(context, info);
844        } catch (Exception e) {
845            synchronized (brokerConnectionStates) {
846                brokerConnectionStates.remove(info.getConnectionId());
847            }
848            unregisterConnectionState(info.getConnectionId());
849            LOG.warn("Failed to add Connection {} due to {}", info.getConnectionId(), e);
850            if (e instanceof SecurityException) {
851                // close this down - in case the peer of this transport doesn't play nice
852                delayedStop(2000, "Failed with SecurityException: " + e.getLocalizedMessage(), e);
853            }
854            throw e;
855        }
856        if (info.isManageable()) {
857            // send ConnectionCommand
858            ConnectionControl command = this.connector.getConnectionControl();
859            command.setFaultTolerant(broker.isFaultTolerantConfiguration());
860            if (info.isFailoverReconnect()) {
861                command.setRebalanceConnection(false);
862            }
863            dispatchAsync(command);
864        }
865        return null;
866    }
867
868    @Override
869    public synchronized Response processRemoveConnection(ConnectionId id, long lastDeliveredSequenceId)
870            throws InterruptedException {
871        LOG.debug("remove connection id: {}", id);
872        TransportConnectionState cs = lookupConnectionState(id);
873        if (cs != null) {
874            // Don't allow things to be added to the connection state while we
875            // are shutting down.
876            cs.shutdown();
877            // Cascade the connection stop to the sessions.
878            for (SessionId sessionId : cs.getSessionIds()) {
879                try {
880                    processRemoveSession(sessionId, lastDeliveredSequenceId);
881                } catch (Throwable e) {
882                    SERVICELOG.warn("Failed to remove session {}", sessionId, e);
883                }
884            }
885            // Cascade the connection stop to temp destinations.
886            for (Iterator<DestinationInfo> iter = cs.getTempDestinations().iterator(); iter.hasNext(); ) {
887                DestinationInfo di = iter.next();
888                try {
889                    broker.removeDestination(cs.getContext(), di.getDestination(), 0);
890                } catch (Throwable e) {
891                    SERVICELOG.warn("Failed to remove tmp destination {}", di.getDestination(), e);
892                }
893                iter.remove();
894            }
895            try {
896                broker.removeConnection(cs.getContext(), cs.getInfo(), transportException.get());
897            } catch (Throwable e) {
898                SERVICELOG.warn("Failed to remove connection {}", cs.getInfo(), e);
899            }
900            TransportConnectionState state = unregisterConnectionState(id);
901            if (state != null) {
902                synchronized (brokerConnectionStates) {
903                    // If we are the last reference, we should remove the state
904                    // from the broker.
905                    if (state.decrementReference() == 0) {
906                        brokerConnectionStates.remove(id);
907                    }
908                }
909            }
910        }
911        return null;
912    }
913
914    @Override
915    public Response processProducerAck(ProducerAck ack) throws Exception {
916        // A broker should not get ProducerAck messages.
917        return null;
918    }
919
920    @Override
921    public Connector getConnector() {
922        return connector;
923    }
924
925    @Override
926    public void dispatchSync(Command message) {
927        try {
928            processDispatch(message);
929        } catch (IOException e) {
930            serviceExceptionAsync(e);
931        }
932    }
933
934    @Override
935    public void dispatchAsync(Command message) {
936        if (!stopping.get()) {
937            if (taskRunner == null) {
938                dispatchSync(message);
939            } else {
940                synchronized (dispatchQueue) {
941                    dispatchQueue.add(message);
942                }
943                try {
944                    taskRunner.wakeup();
945                } catch (InterruptedException e) {
946                    Thread.currentThread().interrupt();
947                }
948            }
949        } else {
950            if (message.isMessageDispatch()) {
951                MessageDispatch md = (MessageDispatch) message;
952                TransmitCallback sub = md.getTransmitCallback();
953                broker.postProcessDispatch(md);
954                if (sub != null) {
955                    sub.onFailure();
956                }
957            }
958        }
959    }
960
961    protected void processDispatch(Command command) throws IOException {
962        MessageDispatch messageDispatch = (MessageDispatch) (command.isMessageDispatch() ? command : null);
963        try {
964            if (!stopping.get()) {
965                if (messageDispatch != null) {
966                    try {
967                        broker.preProcessDispatch(messageDispatch);
968                    } catch (RuntimeException convertToIO) {
969                        throw new IOException(convertToIO);
970                    }
971                }
972                dispatch(command);
973            }
974        } catch (IOException e) {
975            if (messageDispatch != null) {
976                TransmitCallback sub = messageDispatch.getTransmitCallback();
977                broker.postProcessDispatch(messageDispatch);
978                if (sub != null) {
979                    sub.onFailure();
980                }
981                messageDispatch = null;
982                throw e;
983            } else {
984                if (TRANSPORTLOG.isDebugEnabled()) {
985                    TRANSPORTLOG.debug("Unexpected exception on asyncDispatch, command of type: " + command.getDataStructureType(), e);
986                }
987            }
988        } finally {
989            if (messageDispatch != null) {
990                TransmitCallback sub = messageDispatch.getTransmitCallback();
991                broker.postProcessDispatch(messageDispatch);
992                if (sub != null) {
993                    sub.onSuccess();
994                }
995            }
996        }
997    }
998
999    @Override
1000    public boolean iterate() {
1001        try {
1002            if (status.get() == PENDING_STOP || stopping.get()) {
1003                if (dispatchStopped.compareAndSet(false, true)) {
1004                    if (transportException.get() == null) {
1005                        try {
1006                            dispatch(new ShutdownInfo());
1007                        } catch (Throwable ignore) {
1008                        }
1009                    }
1010                    dispatchStoppedLatch.countDown();
1011                }
1012                return false;
1013            }
1014            if (!dispatchStopped.get()) {
1015                Command command = null;
1016                synchronized (dispatchQueue) {
1017                    if (dispatchQueue.isEmpty()) {
1018                        return false;
1019                    }
1020                    command = dispatchQueue.remove(0);
1021                }
1022                processDispatch(command);
1023                return true;
1024            }
1025            return false;
1026        } catch (IOException e) {
1027            if (dispatchStopped.compareAndSet(false, true)) {
1028                dispatchStoppedLatch.countDown();
1029            }
1030            serviceExceptionAsync(e);
1031            return false;
1032        }
1033    }
1034
1035    /**
1036     * Returns the statistics for this connection
1037     */
1038    @Override
1039    public ConnectionStatistics getStatistics() {
1040        return statistics;
1041    }
1042
1043    public MessageAuthorizationPolicy getMessageAuthorizationPolicy() {
1044        return messageAuthorizationPolicy;
1045    }
1046
1047    public void setMessageAuthorizationPolicy(MessageAuthorizationPolicy messageAuthorizationPolicy) {
1048        this.messageAuthorizationPolicy = messageAuthorizationPolicy;
1049    }
1050
1051    @Override
1052    public boolean isManageable() {
1053        return manageable;
1054    }
1055
1056    @Override
1057    public void start() throws Exception {
1058        if (status.compareAndSet(NEW, STARTING)) {
1059            try {
1060                synchronized (this) {
1061                    if (taskRunnerFactory != null) {
1062                        taskRunner = taskRunnerFactory.createTaskRunner(this, "ActiveMQ Connection Dispatcher: "
1063                                + getRemoteAddress());
1064                    } else {
1065                        taskRunner = null;
1066                    }
1067                    transport.start();
1068                    active = true;
1069                    BrokerInfo info = connector.getBrokerInfo().copy();
1070                    if (connector.isUpdateClusterClients()) {
1071                        info.setPeerBrokerInfos(this.broker.getPeerBrokerInfos());
1072                    } else {
1073                        info.setPeerBrokerInfos(null);
1074                    }
1075                    dispatchAsync(info);
1076
1077                    connector.onStarted(this);
1078                }
1079            } catch (Exception e) {
1080                // Force clean up on an error starting up.
1081                status.set(PENDING_STOP);
1082                throw e;
1083            } finally {
1084                // stop() can be called from within the above block,
1085                // but we want to be sure start() completes before
1086                // stop() runs, so queue the stop until right now:
1087                if (!status.compareAndSet(STARTING, STARTED)) {
1088                    LOG.debug("Calling the delayed stop() after start() {}", this);
1089                    stop();
1090                }
1091            }
1092        }
1093    }
1094
1095    @Override
1096    public void stop() throws Exception {
1097        // do not stop task the task runner factories (taskRunnerFactory, stopTaskRunnerFactory)
1098        // as their lifecycle is handled elsewhere
1099
1100        stopAsync();
1101        while (!stopped.await(5, TimeUnit.SECONDS)) {
1102            LOG.info("The connection to '{}' is taking a long time to shutdown.", transport.getRemoteAddress());
1103        }
1104    }
1105
1106    public void delayedStop(final int waitTime, final String reason, Throwable cause) {
1107        if (waitTime > 0) {
1108            status.compareAndSet(STARTING, PENDING_STOP);
1109            transportException.set(cause);
1110            try {
1111                stopTaskRunnerFactory.execute(new Runnable() {
1112                    @Override
1113                    public void run() {
1114                        try {
1115                            Thread.sleep(waitTime);
1116                            stopAsync();
1117                            LOG.info("Stopping {} because {}", transport.getRemoteAddress(), reason);
1118                        } catch (InterruptedException e) {
1119                        }
1120                    }
1121                });
1122            } catch (Throwable t) {
1123                LOG.warn("Cannot create stopAsync. This exception will be ignored.", t);
1124            }
1125        }
1126    }
1127
1128    public void stopAsync(Throwable cause) {
1129        transportException.set(cause);
1130        stopAsync();
1131    }
1132
1133    public void stopAsync() {
1134        // If we're in the middle of starting then go no further... for now.
1135        if (status.compareAndSet(STARTING, PENDING_STOP)) {
1136            LOG.debug("stopAsync() called in the middle of start(). Delaying till start completes..");
1137            return;
1138        }
1139        if (stopping.compareAndSet(false, true)) {
1140            // Let all the connection contexts know we are shutting down
1141            // so that in progress operations can notice and unblock.
1142            List<TransportConnectionState> connectionStates = listConnectionStates();
1143            for (TransportConnectionState cs : connectionStates) {
1144                ConnectionContext connectionContext = cs.getContext();
1145                if (connectionContext != null) {
1146                    connectionContext.getStopping().set(true);
1147                }
1148            }
1149            try {
1150                stopTaskRunnerFactory.execute(new Runnable() {
1151                    @Override
1152                    public void run() {
1153                        serviceLock.writeLock().lock();
1154                        try {
1155                            doStop();
1156                        } catch (Throwable e) {
1157                            LOG.debug("Error occurred while shutting down a connection {}", this, e);
1158                        } finally {
1159                            stopped.countDown();
1160                            serviceLock.writeLock().unlock();
1161                        }
1162                    }
1163                });
1164            } catch (Throwable t) {
1165                LOG.warn("Cannot create async transport stopper thread. This exception is ignored. Not waiting for stop to complete", t);
1166                stopped.countDown();
1167            }
1168        }
1169    }
1170
1171    @Override
1172    public String toString() {
1173        return "Transport Connection to: " + transport.getRemoteAddress();
1174    }
1175
1176    protected void doStop() throws Exception {
1177        LOG.debug("Stopping connection: {}", transport.getRemoteAddress());
1178        connector.onStopped(this);
1179        try {
1180            synchronized (this) {
1181                if (duplexBridge != null) {
1182                    duplexBridge.stop();
1183                }
1184            }
1185        } catch (Exception ignore) {
1186            LOG.trace("Exception caught stopping. This exception is ignored.", ignore);
1187        }
1188        try {
1189            transport.stop();
1190            LOG.debug("Stopped transport: {}", transport.getRemoteAddress());
1191        } catch (Exception e) {
1192            LOG.debug("Could not stop transport to {}. This exception is ignored.", transport.getRemoteAddress(), e);
1193        }
1194        if (taskRunner != null) {
1195            taskRunner.shutdown(1);
1196            taskRunner = null;
1197        }
1198        active = false;
1199        // Run the MessageDispatch callbacks so that message references get
1200        // cleaned up.
1201        synchronized (dispatchQueue) {
1202            for (Iterator<Command> iter = dispatchQueue.iterator(); iter.hasNext(); ) {
1203                Command command = iter.next();
1204                if (command.isMessageDispatch()) {
1205                    MessageDispatch md = (MessageDispatch) command;
1206                    TransmitCallback sub = md.getTransmitCallback();
1207                    broker.postProcessDispatch(md);
1208                    if (sub != null) {
1209                        sub.onFailure();
1210                    }
1211                }
1212            }
1213            dispatchQueue.clear();
1214        }
1215        //
1216        // Remove all logical connection associated with this connection
1217        // from the broker.
1218        if (!broker.isStopped()) {
1219            List<TransportConnectionState> connectionStates = listConnectionStates();
1220            connectionStates = listConnectionStates();
1221            for (TransportConnectionState cs : connectionStates) {
1222                cs.getContext().getStopping().set(true);
1223                try {
1224                    LOG.debug("Cleaning up connection resources: {}", getRemoteAddress());
1225                    processRemoveConnection(cs.getInfo().getConnectionId(), RemoveInfo.LAST_DELIVERED_UNKNOWN);
1226                } catch (Throwable ignore) {
1227                    LOG.debug("Exception caught removing connection {}. This exception is ignored.", cs.getInfo().getConnectionId(), ignore);
1228                }
1229            }
1230        }
1231        LOG.debug("Connection Stopped: {}", getRemoteAddress());
1232    }
1233
1234    /**
1235     * @return Returns the blockedCandidate.
1236     */
1237    public boolean isBlockedCandidate() {
1238        return blockedCandidate;
1239    }
1240
1241    /**
1242     * @param blockedCandidate The blockedCandidate to set.
1243     */
1244    public void setBlockedCandidate(boolean blockedCandidate) {
1245        this.blockedCandidate = blockedCandidate;
1246    }
1247
1248    /**
1249     * @return Returns the markedCandidate.
1250     */
1251    public boolean isMarkedCandidate() {
1252        return markedCandidate;
1253    }
1254
1255    /**
1256     * @param markedCandidate The markedCandidate to set.
1257     */
1258    public void setMarkedCandidate(boolean markedCandidate) {
1259        this.markedCandidate = markedCandidate;
1260        if (!markedCandidate) {
1261            timeStamp = 0;
1262            blockedCandidate = false;
1263        }
1264    }
1265
1266    /**
1267     * @param slow The slow to set.
1268     */
1269    public void setSlow(boolean slow) {
1270        this.slow = slow;
1271    }
1272
1273    /**
1274     * @return true if the Connection is slow
1275     */
1276    @Override
1277    public boolean isSlow() {
1278        return slow;
1279    }
1280
1281    /**
1282     * @return true if the Connection is potentially blocked
1283     */
1284    public boolean isMarkedBlockedCandidate() {
1285        return markedCandidate;
1286    }
1287
1288    /**
1289     * Mark the Connection, so we can deem if it's collectable on the next sweep
1290     */
1291    public void doMark() {
1292        if (timeStamp == 0) {
1293            timeStamp = System.currentTimeMillis();
1294        }
1295    }
1296
1297    /**
1298     * @return if after being marked, the Connection is still writing
1299     */
1300    @Override
1301    public boolean isBlocked() {
1302        return blocked;
1303    }
1304
1305    /**
1306     * @return true if the Connection is connected
1307     */
1308    @Override
1309    public boolean isConnected() {
1310        return connected;
1311    }
1312
1313    /**
1314     * @param blocked The blocked to set.
1315     */
1316    public void setBlocked(boolean blocked) {
1317        this.blocked = blocked;
1318    }
1319
1320    /**
1321     * @param connected The connected to set.
1322     */
1323    public void setConnected(boolean connected) {
1324        this.connected = connected;
1325    }
1326
1327    /**
1328     * @return true if the Connection is active
1329     */
1330    @Override
1331    public boolean isActive() {
1332        return active;
1333    }
1334
1335    /**
1336     * @param active The active to set.
1337     */
1338    public void setActive(boolean active) {
1339        this.active = active;
1340    }
1341
1342    /**
1343     * @return true if the Connection is starting
1344     */
1345    public boolean isStarting() {
1346        return status.get() == STARTING;
1347    }
1348
1349    @Override
1350    public synchronized boolean isNetworkConnection() {
1351        return networkConnection;
1352    }
1353
1354    @Override
1355    public boolean isFaultTolerantConnection() {
1356        return this.faultTolerantConnection;
1357    }
1358
1359    /**
1360     * @return true if the Connection needs to stop
1361     */
1362    public boolean isPendingStop() {
1363        return status.get() == PENDING_STOP;
1364    }
1365
1366    @Override
1367    public Response processBrokerInfo(BrokerInfo info) {
1368        if (info.isSlaveBroker()) {
1369            LOG.error(" Slave Brokers are no longer supported - slave trying to attach is: {}", info.getBrokerName());
1370        } else if (info.isNetworkConnection() && info.isDuplexConnection()) {
1371            // so this TransportConnection is the rear end of a network bridge
1372            // We have been requested to create a two way pipe ...
1373            try {
1374                Properties properties = MarshallingSupport.stringToProperties(info.getNetworkProperties());
1375                Map<String, String> props = createMap(properties);
1376                NetworkBridgeConfiguration config = new NetworkBridgeConfiguration();
1377                IntrospectionSupport.setProperties(config, props, "");
1378                config.setBrokerName(broker.getBrokerName());
1379
1380                // check for existing duplex connection hanging about
1381
1382                // We first look if existing network connection already exists for the same broker Id and network connector name
1383                // It's possible in case of brief network fault to have this transport connector side of the connection always active
1384                // and the duplex network connector side wanting to open a new one
1385                // In this case, the old connection must be broken
1386                String duplexNetworkConnectorId = config.getName() + "@" + info.getBrokerId();
1387                CopyOnWriteArrayList<TransportConnection> connections = this.connector.getConnections();
1388                synchronized (connections) {
1389                    for (Iterator<TransportConnection> iter = connections.iterator(); iter.hasNext(); ) {
1390                        TransportConnection c = iter.next();
1391                        if ((c != this) && (duplexNetworkConnectorId.equals(c.getDuplexNetworkConnectorId()))) {
1392                            LOG.warn("Stopping an existing active duplex connection [{}] for network connector ({}).", c, duplexNetworkConnectorId);
1393                            c.stopAsync();
1394                            // better to wait for a bit rather than get connection id already in use and failure to start new bridge
1395                            c.getStopped().await(1, TimeUnit.SECONDS);
1396                        }
1397                    }
1398                    setDuplexNetworkConnectorId(duplexNetworkConnectorId);
1399                }
1400                Transport localTransport = NetworkBridgeFactory.createLocalTransport(config, broker.getVmConnectorURI());
1401                Transport remoteBridgeTransport = transport;
1402                if (! (remoteBridgeTransport instanceof ResponseCorrelator)) {
1403                    // the vm transport case is already wrapped
1404                    remoteBridgeTransport = new ResponseCorrelator(remoteBridgeTransport);
1405                }
1406                String duplexName = localTransport.toString();
1407                if (duplexName.contains("#")) {
1408                    duplexName = duplexName.substring(duplexName.lastIndexOf("#"));
1409                }
1410                MBeanNetworkListener listener = new MBeanNetworkListener(brokerService, config, brokerService.createDuplexNetworkConnectorObjectName(duplexName));
1411                listener.setCreatedByDuplex(true);
1412                duplexBridge = NetworkBridgeFactory.createBridge(config, localTransport, remoteBridgeTransport, listener);
1413                duplexBridge.setBrokerService(brokerService);
1414                // now turn duplex off this side
1415                info.setDuplexConnection(false);
1416                duplexBridge.setCreatedByDuplex(true);
1417                duplexBridge.duplexStart(this, brokerInfo, info);
1418                LOG.info("Started responder end of duplex bridge {}", duplexNetworkConnectorId);
1419                return null;
1420            } catch (TransportDisposedIOException e) {
1421                LOG.warn("Duplex bridge {} was stopped before it was correctly started.", duplexNetworkConnectorId);
1422                return null;
1423            } catch (Exception e) {
1424                LOG.error("Failed to create responder end of duplex network bridge {}", duplexNetworkConnectorId, e);
1425                return null;
1426            }
1427        }
1428        // We only expect to get one broker info command per connection
1429        if (this.brokerInfo != null) {
1430            LOG.warn("Unexpected extra broker info command received: {}", info);
1431        }
1432        this.brokerInfo = info;
1433        networkConnection = true;
1434        List<TransportConnectionState> connectionStates = listConnectionStates();
1435        for (TransportConnectionState cs : connectionStates) {
1436            cs.getContext().setNetworkConnection(true);
1437        }
1438        return null;
1439    }
1440
1441    @SuppressWarnings({"unchecked", "rawtypes"})
1442    private HashMap<String, String> createMap(Properties properties) {
1443        return new HashMap(properties);
1444    }
1445
1446    protected void dispatch(Command command) throws IOException {
1447        try {
1448            setMarkedCandidate(true);
1449            transport.oneway(command);
1450        } finally {
1451            setMarkedCandidate(false);
1452        }
1453    }
1454
1455    @Override
1456    public String getRemoteAddress() {
1457        return transport.getRemoteAddress();
1458    }
1459
1460    public Transport getTransport() {
1461        return transport;
1462    }
1463
1464    @Override
1465    public String getConnectionId() {
1466        List<TransportConnectionState> connectionStates = listConnectionStates();
1467        for (TransportConnectionState cs : connectionStates) {
1468            if (cs.getInfo().getClientId() != null) {
1469                return cs.getInfo().getClientId();
1470            }
1471            return cs.getInfo().getConnectionId().toString();
1472        }
1473        return null;
1474    }
1475
1476    @Override
1477    public void updateClient(ConnectionControl control) {
1478        if (isActive() && isBlocked() == false && isFaultTolerantConnection() && this.wireFormatInfo != null
1479                && this.wireFormatInfo.getVersion() >= 6) {
1480            dispatchAsync(control);
1481        }
1482    }
1483
1484    public ProducerBrokerExchange getProducerBrokerExchangeIfExists(ProducerInfo producerInfo){
1485        ProducerBrokerExchange result = null;
1486        if (producerInfo != null && producerInfo.getProducerId() != null){
1487            synchronized (producerExchanges){
1488                result = producerExchanges.get(producerInfo.getProducerId());
1489            }
1490        }
1491        return result;
1492    }
1493
1494    private ProducerBrokerExchange getProducerBrokerExchange(ProducerId id) throws IOException {
1495        ProducerBrokerExchange result = producerExchanges.get(id);
1496        if (result == null) {
1497            synchronized (producerExchanges) {
1498                result = new ProducerBrokerExchange();
1499                TransportConnectionState state = lookupConnectionState(id);
1500                context = state.getContext();
1501                result.setConnectionContext(context);
1502                if (context.isReconnect() || (context.isNetworkConnection() && connector.isAuditNetworkProducers())) {
1503                    result.setLastStoredSequenceId(brokerService.getPersistenceAdapter().getLastProducerSequenceId(id));
1504                }
1505                SessionState ss = state.getSessionState(id.getParentId());
1506                if (ss != null) {
1507                    result.setProducerState(ss.getProducerState(id));
1508                    ProducerState producerState = ss.getProducerState(id);
1509                    if (producerState != null && producerState.getInfo() != null) {
1510                        ProducerInfo info = producerState.getInfo();
1511                        result.setMutable(info.getDestination() == null || info.getDestination().isComposite());
1512                    }
1513                }
1514                producerExchanges.put(id, result);
1515            }
1516        } else {
1517            context = result.getConnectionContext();
1518        }
1519        return result;
1520    }
1521
1522    private void removeProducerBrokerExchange(ProducerId id) {
1523        synchronized (producerExchanges) {
1524            producerExchanges.remove(id);
1525        }
1526    }
1527
1528    private ConsumerBrokerExchange getConsumerBrokerExchange(ConsumerId id) {
1529        ConsumerBrokerExchange result = consumerExchanges.get(id);
1530        return result;
1531    }
1532
1533    private ConsumerBrokerExchange addConsumerBrokerExchange(TransportConnectionState connectionState, ConsumerId id) {
1534        ConsumerBrokerExchange result = consumerExchanges.get(id);
1535        if (result == null) {
1536            synchronized (consumerExchanges) {
1537                result = new ConsumerBrokerExchange();
1538                context = connectionState.getContext();
1539                result.setConnectionContext(context);
1540                SessionState ss = connectionState.getSessionState(id.getParentId());
1541                if (ss != null) {
1542                    ConsumerState cs = ss.getConsumerState(id);
1543                    if (cs != null) {
1544                        ConsumerInfo info = cs.getInfo();
1545                        if (info != null) {
1546                            if (info.getDestination() != null && info.getDestination().isPattern()) {
1547                                result.setWildcard(true);
1548                            }
1549                        }
1550                    }
1551                }
1552                consumerExchanges.put(id, result);
1553            }
1554        }
1555        return result;
1556    }
1557
1558    private void removeConsumerBrokerExchange(ConsumerId id) {
1559        synchronized (consumerExchanges) {
1560            consumerExchanges.remove(id);
1561        }
1562    }
1563
1564    public int getProtocolVersion() {
1565        return protocolVersion.get();
1566    }
1567
1568    @Override
1569    public Response processControlCommand(ControlCommand command) throws Exception {
1570        return null;
1571    }
1572
1573    @Override
1574    public Response processMessageDispatch(MessageDispatch dispatch) throws Exception {
1575        return null;
1576    }
1577
1578    @Override
1579    public Response processConnectionControl(ConnectionControl control) throws Exception {
1580        if (control != null) {
1581            faultTolerantConnection = control.isFaultTolerant();
1582        }
1583        return null;
1584    }
1585
1586    @Override
1587    public Response processConnectionError(ConnectionError error) throws Exception {
1588        return null;
1589    }
1590
1591    @Override
1592    public Response processConsumerControl(ConsumerControl control) throws Exception {
1593        ConsumerBrokerExchange consumerExchange = getConsumerBrokerExchange(control.getConsumerId());
1594        broker.processConsumerControl(consumerExchange, control);
1595        return null;
1596    }
1597
1598    protected synchronized TransportConnectionState registerConnectionState(ConnectionId connectionId,
1599                                                                            TransportConnectionState state) {
1600        TransportConnectionState cs = null;
1601        if (!connectionStateRegister.isEmpty() && !connectionStateRegister.doesHandleMultipleConnectionStates()) {
1602            // swap implementations
1603            TransportConnectionStateRegister newRegister = new MapTransportConnectionStateRegister();
1604            newRegister.intialize(connectionStateRegister);
1605            connectionStateRegister = newRegister;
1606        }
1607        cs = connectionStateRegister.registerConnectionState(connectionId, state);
1608        return cs;
1609    }
1610
1611    protected synchronized TransportConnectionState unregisterConnectionState(ConnectionId connectionId) {
1612        return connectionStateRegister.unregisterConnectionState(connectionId);
1613    }
1614
1615    protected synchronized List<TransportConnectionState> listConnectionStates() {
1616        return connectionStateRegister.listConnectionStates();
1617    }
1618
1619    protected synchronized TransportConnectionState lookupConnectionState(String connectionId) {
1620        return connectionStateRegister.lookupConnectionState(connectionId);
1621    }
1622
1623    protected synchronized TransportConnectionState lookupConnectionState(ConsumerId id) {
1624        return connectionStateRegister.lookupConnectionState(id);
1625    }
1626
1627    protected synchronized TransportConnectionState lookupConnectionState(ProducerId id) {
1628        return connectionStateRegister.lookupConnectionState(id);
1629    }
1630
1631    protected synchronized TransportConnectionState lookupConnectionState(SessionId id) {
1632        return connectionStateRegister.lookupConnectionState(id);
1633    }
1634
1635    // public only for testing
1636    public synchronized TransportConnectionState lookupConnectionState(ConnectionId connectionId) {
1637        return connectionStateRegister.lookupConnectionState(connectionId);
1638    }
1639
1640    protected synchronized void setDuplexNetworkConnectorId(String duplexNetworkConnectorId) {
1641        this.duplexNetworkConnectorId = duplexNetworkConnectorId;
1642    }
1643
1644    protected synchronized String getDuplexNetworkConnectorId() {
1645        return this.duplexNetworkConnectorId;
1646    }
1647
1648    public boolean isStopping() {
1649        return stopping.get();
1650    }
1651
1652    protected CountDownLatch getStopped() {
1653        return stopped;
1654    }
1655
1656    private int getProducerCount(ConnectionId connectionId) {
1657        int result = 0;
1658        TransportConnectionState cs = lookupConnectionState(connectionId);
1659        if (cs != null) {
1660            for (SessionId sessionId : cs.getSessionIds()) {
1661                SessionState sessionState = cs.getSessionState(sessionId);
1662                if (sessionState != null) {
1663                    result += sessionState.getProducerIds().size();
1664                }
1665            }
1666        }
1667        return result;
1668    }
1669
1670    private int getConsumerCount(ConnectionId connectionId) {
1671        int result = 0;
1672        TransportConnectionState cs = lookupConnectionState(connectionId);
1673        if (cs != null) {
1674            for (SessionId sessionId : cs.getSessionIds()) {
1675                SessionState sessionState = cs.getSessionState(sessionId);
1676                if (sessionState != null) {
1677                    result += sessionState.getConsumerIds().size();
1678                }
1679            }
1680        }
1681        return result;
1682    }
1683
1684    public WireFormatInfo getRemoteWireFormatInfo() {
1685        return wireFormatInfo;
1686    }
1687}