import org.apache.activemq.transport.TransportListener; //导入依赖的package包/类
@Override
protected void doStart() throws Exception {
    LOG.info("Starting " + this);

    configuredTransport.setTransportListener(new TransportListener() {
        @Override
        public void onCommand(Object o) {
            final Command command = (Command)o;
            processInboundConnection(command);
        }

        @Override
        public void onException(IOException error) {
            LOG.error("Caught: " + error, error);
        }

        @Override
        public void transportInterupted() {
        }

        @Override
        public void transportResumed() {
        }
    });
    configuredTransport.start();
}
 

import org.apache.activemq.transport.TransportListener; //导入依赖的package包/类
public void dispatch(VMTransport transport, BlockingQueue<Object> pending, Object command) {
    TransportListener transportListener = transport.getTransportListener();
    if (transportListener != null) {
        // Lock here on the target transport's started since we want to wait for its start()
        // method to finish dispatching out of the queue before we do our own.
        synchronized (transport.started) {

            // Ensure that no additional commands entered the queue in the small time window
            // before the start method locks the dispatch lock and the oneway method was in
            // an put operation.
            while(pending != null && !pending.isEmpty() && !transport.isDisposed()) {
                doDispatch(transport, transportListener, pending.poll());
            }

            // We are now in sync mode and won't enqueue any more commands to the target
            // transport so lets clean up its resources.
            transport.messageQueue = null;

            // Don't dispatch if either end was disposed already.
            if (command != null && !this.disposed.get() && !transport.isDisposed()) {
                doDispatch(transport, transportListener, command);
            }
        }
    }
}
 

import org.apache.activemq.transport.TransportListener; //导入依赖的package包/类
@Override
public void onException(final IOException error) {
    onAsyncException(error);
    if (!closing.get() && !closed.get()) {
        executor.execute(new Runnable() {
            @Override
            public void run() {
                transportFailed(error);
                ServiceSupport.dispose(ActiveMQConnection.this.transport);
                brokerInfoReceived.countDown();
                try {
                    cleanup();
                } catch (JMSException e) {
                    LOG.warn("Exception during connection cleanup, " + e, e);
                }
                for (Iterator<TransportListener> iter = transportListeners.iterator(); iter.hasNext();) {
                    TransportListener listener = iter.next();
                    listener.onException(error);
                }
            }
        });
    }
}
 

import org.apache.activemq.transport.TransportListener; //导入依赖的package包/类
@Override
public void transportInterupted() {
    transportInterruptionProcessingComplete.set(1);
    for (Iterator<ActiveMQSession> i = this.sessions.iterator(); i.hasNext();) {
        ActiveMQSession s = i.next();
        s.clearMessagesInProgress(transportInterruptionProcessingComplete);
    }

    for (ActiveMQConnectionConsumer connectionConsumer : this.connectionConsumers) {
        connectionConsumer.clearMessagesInProgress(transportInterruptionProcessingComplete);
    }

    if (transportInterruptionProcessingComplete.decrementAndGet() > 0) {
        if (LOG.isDebugEnabled()) {
            LOG.debug("transport interrupted - processing required, dispatchers: " + transportInterruptionProcessingComplete.get());
        }
        signalInterruptionProcessingNeeded();
    }

    for (Iterator<TransportListener> iter = transportListeners.iterator(); iter.hasNext();) {
        TransportListener listener = iter.next();
        listener.transportInterupted();
    }
}
 

import org.apache.activemq.transport.TransportListener; //导入依赖的package包/类
protected Transport createTransport() throws Exception {
   Transport transport = TransportFactory.connect(new URI("failover://(tcp://localhost:1234?transport.connectTimeout=10000)"));
   transport.setTransportListener(new TransportListener() {

      @Override
      public void onCommand(Object command) {
      }

      @Override
      public void onException(IOException error) {
      }

      @Override
      public void transportInterupted() {
      }

      @Override
      public void transportResumed() {
      }
   });
   transport.start();

   this.failoverTransport = transport.narrow(FailoverTransport.class);

   return transport;
}
 

import org.apache.activemq.transport.TransportListener; //导入依赖的package包/类
public void doDispatch(VMTransport transport, TransportListener transportListener, Object command) {
    if (command == DISCONNECT) {
        transportListener.onException(new TransportDisposedIOException("Peer (" + peer.toString() + ") disposed."));
    } else {
        transport.receiveCounter++;
        transportListener.onCommand(command);
    }
}
 

import org.apache.activemq.transport.TransportListener; //导入依赖的package包/类
/**
 * @see org.apache.activemq.thread.Task#iterate()
 */
public boolean iterate() {

    final TransportListener tl = transportListener;

    LinkedBlockingQueue<Object> mq;
    try {
        mq = getMessageQueue();
    } catch (TransportDisposedIOException e) {
        return false;
    }

    Object command = mq.poll();
    if (command != null && !disposed.get()) {
        if( command == DISCONNECT ) {
            tl.onException(new TransportDisposedIOException("Peer (" + peer.toString() + ") disposed."));
        } else {
            tl.onCommand(command);
        }
        return !mq.isEmpty() && !disposed.get();
    } else {
        if(disposed.get()) {
            mq.clear();
        }
        return false;
    }
}
 

import org.apache.activemq.transport.TransportListener; //导入依赖的package包/类
/**
 */
public synchronized void setTransportListener(TransportListener channelListener) {
    this.transportListener = channelListener;
    if (channelListener == null) {
        getNext().setTransportListener(null);
    } else {
        getNext().setTransportListener(this);
    }
}
 

import org.apache.activemq.transport.TransportListener; //导入依赖的package包/类
@Override
public void sendToActiveMQ(Command command) {
    TransportListener l = transportListener;
    if (l != null) {
        l.onCommand(command);
    }
}
 

import org.apache.activemq.transport.TransportListener; //导入依赖的package包/类
@Override
public void setTransportListener(TransportListener commandListener) {
    synchronized (listenerMutex) {
        this.transportListener = commandListener;
        listenerMutex.notifyAll();
    }
}
 

import org.apache.activemq.transport.TransportListener; //导入依赖的package包/类
@Override
public void transportResumed() {
    for (Iterator<TransportListener> iter = transportListeners.iterator(); iter.hasNext();) {
        TransportListener listener = iter.next();
        listener.transportResumed();
    }
}
 

import org.apache.activemq.transport.TransportListener; //导入依赖的package包/类
/**
 * @throws Exception
 * @throws URISyntaxException
 */
private void startClient(String uri) throws Exception, URISyntaxException {
   clientTransport = TransportFactory.connect(new URI(uri));
   clientTransport.setTransportListener(new TransportListener() {
      @Override
      public void onCommand(Object command) {
         if (command instanceof WireFormatInfo) {
            clientWF.set((WireFormatInfo) command);
            negotiationCounter.countDown();
         }
      }

      @Override
      public void onException(IOException error) {
         if (!ignoreAsycError.get()) {
            LOG.info("Client transport error: ", error);
            asyncError.set(error);
            negotiationCounter.countDown();
         }
      }

      @Override
      public void transportInterupted() {
      }

      @Override
      public void transportResumed() {
      }
   });
   clientTransport.start();
}
 

import org.apache.activemq.transport.TransportListener; //导入依赖的package包/类
/**
 * @throws Exception
 * @throws URISyntaxException
 */
private void startClient() throws Exception, URISyntaxException {
   clientTransport = TransportFactory.connect(new URI("tcp://localhost:" + serverPort + "?trace=true&wireFormat.maxInactivityDuration=1000"));
   clientTransport.setTransportListener(new TransportListener() {
      @Override
      public void onCommand(Object command) {
         clientReceiveCount.incrementAndGet();
         if (clientRunOnCommand != null) {
            clientRunOnCommand.run();
         }
      }

      @Override
      public void onException(IOException error) {
         if (!ignoreClientError.get()) {
            LOG.info("Client transport error:");
            error.printStackTrace();
            clientErrorCount.incrementAndGet();
         }
      }

      @Override
      public void transportInterupted() {
      }

      @Override
      public void transportResumed() {
      }
   });
   clientTransport.start();
}
 

import org.apache.activemq.transport.TransportListener; //导入依赖的package包/类
@Override
public void onAccept(Transport transport) {
   try {
      LOG.info("[" + getName() + "] Server Accepted a Connection");
      serverTransport = transport;
      serverTransport.setTransportListener(new TransportListener() {
         @Override
         public void onCommand(Object command) {
            serverReceiveCount.incrementAndGet();
            if (serverRunOnCommand != null) {
               serverRunOnCommand.run();
            }
         }

         @Override
         public void onException(IOException error) {
            if (!ignoreClientError.get()) {
               LOG.info("Server transport error:", error);
               serverErrorCount.incrementAndGet();
            }
         }

         @Override
         public void transportInterupted() {
         }

         @Override
         public void transportResumed() {
         }
      });
      serverTransport.start();
   } catch (Exception e) {
      e.printStackTrace();
   }
}
 

import org.apache.activemq.transport.TransportListener; //导入依赖的package包/类
public void testNoBrokersInBrokerInfo() throws Exception {
   final BrokerInfo info[] = new BrokerInfo[1];
   TransportListener listener = new TransportListener() {
      @Override
      public void onCommand(Object command) {
         LOG.info("Got command: " + command);
         if (command instanceof BrokerInfo) {
            info[0] = (BrokerInfo) command;
         }
      }

      @Override
      public void onException(IOException error) {
         //To change body of implemented methods use File | Settings | File Templates.
      }

      @Override
      public void transportInterupted() {
         //To change body of implemented methods use File | Settings | File Templates.
      }

      @Override
      public void transportResumed() {
         //To change body of implemented methods use File | Settings | File Templates.
      }
   };
   @SuppressWarnings("unused")
   StubConnection c = createFailoverConnection(listener);
   int count = 0;
   while (count++ < 20 && info[0] == null) {
      TimeUnit.SECONDS.sleep(1);
   }
   Assert.assertNotNull("got a valid brokerInfo after 20 secs", info[0]);
   Assert.assertNull("no peer brokers present", info[0].getPeerBrokerInfos());
}
 

import org.apache.activemq.transport.TransportListener; //导入依赖的package包/类
protected StubConnection createFailoverConnection(TransportListener listener) throws Exception {
   URI failoverURI = new URI("failover://" + newURI(0) + "," + newURI(1) + "");
   Transport transport = TransportFactory.connect(failoverURI);
   StubConnection connection = new StubConnection(transport, listener);
   connections.add(connection);
   return connection;
}
 

import org.apache.activemq.transport.TransportListener; //导入依赖的package包/类
public Worker(final String name) throws URISyntaxException, JMSException {
   this.name = name;
   URI uri = new URI("failover://(mock://(" + tcpUri + "))?updateURIsSupported=false");
   ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(uri);
   connection = (ActiveMQConnection) factory.createConnection();
   connection.addTransportListener(new TransportListener() {
      @Override
      public void onCommand(Object command) {
      }

      @Override
      public void onException(IOException error) {
         setError(error);
      }

      @Override
      public void transportInterupted() {
         LOG.info("Worker " + name + " was interrupted...");
         interruptedCount.incrementAndGet();
      }

      @Override
      public void transportResumed() {
         LOG.info("Worker " + name + " was resummed...");
         resumedCount.incrementAndGet();
      }
   });
   connection.start();
}
 

import org.apache.activemq.transport.TransportListener; //导入依赖的package包/类
@Test(timeout = 30000)
@Ignore("Test fails on windows")
public void testReconnectUnlimited() throws Exception {

   Transport transport = TransportFactory.connect(new URI("failover://(tcp://0.0.0.0:61616)?useExponentialBackOff=false&reconnectDelay=0&initialReconnectDelay=0"));

   transport.setTransportListener(new TransportListener() {

      @Override
      public void onCommand(Object command) {
      }

      @Override
      public void onException(IOException error) {
      }

      @Override
      public void transportInterupted() {
      }

      @Override
      public void transportResumed() {
      }
   });
   transport.start();

   this.failoverTransport = transport.narrow(FailoverTransport.class);

   assertTrue("no implicit limit of 1000", Wait.waitFor(new Wait.Condition() {
      @Override
      public boolean isSatisified() throws Exception {
         return failoverTransport.getConnectFailures() > 1002;
      }
   }));
}
 

import org.apache.activemq.transport.TransportListener; //导入依赖的package包/类
@Test
public void testLocalhostPortSyntax() throws Exception {
   transport = TransportFactory.connect(new URI("failover://(tcp://localhost:1111/localhost:2111)"));

   transport.setTransportListener(new TransportListener() {

      @Override
      public void onCommand(Object command) {
      }

      @Override
      public void onException(IOException error) {
      }

      @Override
      public void transportInterupted() {
      }

      @Override
      public void transportResumed() {
      }
   });

   failoverTransport = transport.narrow(FailoverTransport.class);

   transport.start();

}
 

import org.apache.activemq.transport.TransportListener; //导入依赖的package包/类
/**
 * Get the current connection or create one using the connectionSupplier.
 *
 * @return JMS Connection
 */
private Connection getConnection() {

    if (!connectionOption.isPresent()) {
        final Connection connection = connectionSupplier.get();

        if (connection instanceof ActiveMQConnection) {
            ((ActiveMQConnection) connection).addTransportListener(new TransportListener() {
                @Override
                public void onCommand(Object command) {

                }

                @Override
                public void onException(IOException error) {
                }

                @Override
                public void transportInterupted() {
                    connected.set(false);
                }

                @Override
                public void transportResumed() {
                    connected.set(true);
                }
            });
        }


        connected.set(true);

        if (startConnection) {
            try {
                connection.start();
            } catch (JMSException e) {
                throw new JmsException("Unable to start JMS connection", e);
            }
        }
        connectionOption = Optional.of(connection);
    }
    return connectionOption.get();
}