import org.apache.qpid.proton.engine.EndpointState; //导入依赖的package包/类
@Override
public void bind(Connection conn)
{
    // TODO - check if already bound

    _connectionEndpoint = (ConnectionImpl) conn;
    put(Event.Type.CONNECTION_BOUND, conn);
    _connectionEndpoint.setTransport(this);
    _connectionEndpoint.incref();

    if(getRemoteState() != EndpointState.UNINITIALIZED)
    {
        _connectionEndpoint.handleOpen(_open);
        if(getRemoteState() == EndpointState.CLOSED)
        {
            _connectionEndpoint.setRemoteState(EndpointState.CLOSED);
        }

        _frameParser.flush();
    }
}
 

import org.apache.qpid.proton.engine.EndpointState; //导入依赖的package包/类
@Override
public void handleEnd(End end, Binary payload, Integer channel)
{
    TransportSession transportSession = _remoteSessions.get(channel);
    if(transportSession == null)
    {
        // TODO - fail due to attach on non-begun session
    }
    else
    {
        _remoteSessions.remove(channel);
        transportSession.receivedEnd();
        transportSession.unsetRemoteChannel();
        SessionImpl session = transportSession.getSession();
        session.setRemoteState(EndpointState.CLOSED);
        ErrorCondition errorCondition = end.getError();
        if(errorCondition != null)
        {
            session.getRemoteCondition().copyFrom(errorCondition);
        }

        _connectionEndpoint.put(Event.Type.SESSION_REMOTE_CLOSE, session);
    }
}
 

import org.apache.qpid.proton.engine.EndpointState; //导入依赖的package包/类
@Override
public void handleClose(Close close, Binary payload, Integer channel)
{
    _closeReceived = true;
    _remoteIdleTimeout = 0;
    setRemoteState(EndpointState.CLOSED);
    if(_connectionEndpoint != null)
    {
        _connectionEndpoint.setRemoteState(EndpointState.CLOSED);
        if(close.getError() != null)
        {
            _connectionEndpoint.getRemoteCondition().copyFrom(close.getError());
        }

        _connectionEndpoint.put(Event.Type.CONNECTION_REMOTE_CLOSE, _connectionEndpoint);
    }

}
 

import org.apache.qpid.proton.engine.EndpointState; //导入依赖的package包/类
@Override
public int send(final byte[] bytes, int offset, int length)
{
    if (getLocalState() == EndpointState.CLOSED)
    {
        throw new IllegalStateException("send not allowed after the sender is closed.");
    }
    DeliveryImpl current = current();
    if (current == null || current.getLink() != this)
    {
        throw new IllegalArgumentException();//TODO.
    }
    int sent = current.send(bytes, offset, length);
    if (sent > 0) {
        getSession().incrementOutgoingBytes(sent);
    }
    return sent;
}
 

import org.apache.qpid.proton.engine.EndpointState; //导入依赖的package包/类
@Test
public void test()
{
    Connection connection1 = Proton.connection();
    Connection connection2 = Proton.connection();;
    Transport transport1 = Proton.transport();
    transport1.bind(connection1);

    Transport transport2 = Proton.transport();
    transport2.bind(connection2);

    assertEquals(EndpointState.UNINITIALIZED, connection1.getLocalState());
    assertEquals(EndpointState.UNINITIALIZED, connection1.getRemoteState());

    connection1.open();
    connection2.open();
}
 

import org.apache.qpid.proton.engine.EndpointState; //导入依赖的package包/类
private void processEventSessionRemoteState(Event event) {
    final String methodName = "processEventSessionRemoteState";
    logger.entry(this, methodName, event);

    if (event.getSession().getRemoteState() == EndpointState.ACTIVE) {
        if (event.getSession().getLocalState() == EndpointState.ACTIVE) {
            final EngineConnection engineConnection =
                    (EngineConnection) event.getConnection().getContext();
            if (!engineConnection.closed) {
                // First session has opened on the connection
                OpenRequest req = engineConnection.openRequest;
                engineConnection.openRequest = null;
                engineConnection.requestor.tell(new OpenResponse(req, engineConnection), this);
            }
        } else {
            // The remote end is trying to establish a new session with us, which is not allowed. I don't think this is a usual case,
            // but could occur with a badly written remote end (i.e. sends an initial AMQP open immediately followed by a begin)
            final Connection protonConnection = event.getConnection();
            protonConnection.setCondition(new ErrorCondition(Symbol.getSymbol("mqlight:session-remote-open-rejected"),
                                                             "MQ Light client is unable to accept an open session request"));
            protonConnection.close();
        }
    }

    logger.exit(this, methodName);
}
 

import org.apache.qpid.proton.engine.EndpointState; //导入依赖的package包/类
@Override
public void detach(AsyncResult request) {
   // If already closed signal success or else the caller might never get notified.
   if (getEndpoint().getLocalState() == EndpointState.CLOSED || getEndpoint().getRemoteState() == EndpointState.CLOSED) {

      if (getEndpoint().getLocalState() != EndpointState.CLOSED) {
         doDetach();
         getEndpoint().free();
      }

      request.onSuccess();
   } else {
      this.closeRequest = request;
      doDetach();
   }
}
 

import org.apache.qpid.proton.engine.EndpointState; //导入依赖的package包/类
@Override
public void close(AsyncResult request) {
   // If already closed signal success or else the caller might never get notified.
   if (getEndpoint().getLocalState() == EndpointState.CLOSED || getEndpoint().getRemoteState() == EndpointState.CLOSED) {

      if (getEndpoint().getLocalState() != EndpointState.CLOSED) {
         doClose();
         getEndpoint().free();
      }

      request.onSuccess();
   } else {
      this.closeRequest = request;
      doClose();
   }
}
 

import org.apache.qpid.proton.engine.EndpointState; //导入依赖的package包/类
public long tick(boolean firstTick) {
   lock.lock();
   try {
      if (!firstTick) {
         try {
            if (connection.getLocalState() != EndpointState.CLOSED) {
               long rescheduleAt = transport.tick(TimeUnit.NANOSECONDS.toMillis(System.nanoTime()));
               if (transport.isClosed()) {
                  throw new IllegalStateException("Channel was inactive for to long");
               }
               return rescheduleAt;
            }
         } catch (Exception e) {
            log.warn(e.getMessage(), e);
            transport.close();
            connection.setCondition(new ErrorCondition());
         }
         return 0;
      }
      return transport.tick(TimeUnit.NANOSECONDS.toMillis(System.nanoTime()));
   } finally {
      lock.unlock();
      flushBytes();
   }
}
 

import org.apache.qpid.proton.engine.EndpointState; //导入依赖的package包/类
@Override
public void processStateChange() throws IOException {
    EndpointState remoteState = endpoint.getRemoteState();

    if (remoteState == EndpointState.ACTIVE) {
        if (isAwaitingOpen()) {
            LOG.debug("Link {} is now open: ", this);
            opened();
        }

        // Should not receive an ACTIVE event if not awaiting the open state.
    } else if (remoteState == EndpointState.CLOSED) {
        if (isAwaitingClose()) {
            LOG.debug("Link {} is now closed: ", this);
            closed();
        } else if (isAwaitingOpen()) {
            // Error on Open, create exception and signal failure.
            LOG.warn("Open of link {} failed: ", this);
            Exception remoteError = this.getRemoteError();
            failed(remoteError);
        } else {
            // TODO - Handle remote asynchronous close.
        }
    }
}
 

import org.apache.qpid.proton.engine.EndpointState; //导入依赖的package包/类
private void performConnectionOpen(Connection connection, InputStream in, OutputStream out) throws IOException, LoginException {
    connection.setHostname(saslHostname);
    connection.setContainer(container);
    connection.open();
    writeToNetwork(connection, out);
    readFromNetwork(connection, in, () -> connection.getRemoteState() == EndpointState.UNINITIALIZED);
}
 

import org.apache.qpid.proton.engine.EndpointState; //导入依赖的package包/类
@Override
public void onConnectionRemoteOpen(Event evt) {
    Connection conn = evt.getConnection();
    if (conn.getLocalState() == EndpointState.UNINITIALIZED) {
        conn.open();
    }
}
 

import org.apache.qpid.proton.engine.EndpointState; //导入依赖的package包/类
@Override
public void onSessionRemoteOpen(Event evt) {
    Session ssn = evt.getSession();
    if (ssn.getLocalState() == EndpointState.UNINITIALIZED) {
        ssn.open();
    }
}
 

import org.apache.qpid.proton.engine.EndpointState; //导入依赖的package包/类
@Override
public void onLinkRemoteOpen(Event evt) {
    Link link = evt.getLink();
    if (link.getLocalState() == EndpointState.UNINITIALIZED) {
        link.setSource(link.getRemoteSource());
        link.setTarget(link.getRemoteTarget());
        link.open();
    }
}
 

import org.apache.qpid.proton.engine.EndpointState; //导入依赖的package包/类
@Override
public void onConnectionRemoteClose(Event evt) {
    Connection conn = evt.getConnection();
    if (conn.getLocalState() != EndpointState.CLOSED) {
        conn.close();
    }
}
 

import org.apache.qpid.proton.engine.EndpointState; //导入依赖的package包/类
@Override
public void onSessionRemoteClose(Event evt) {
    Session ssn = evt.getSession();
    if (ssn.getLocalState() != EndpointState.CLOSED) {
        ssn.close();
    }
}
 

import org.apache.qpid.proton.engine.EndpointState; //导入依赖的package包/类
@Override
public void onLinkRemoteClose(Event evt) {
    Link link = evt.getLink();
    if (link.getLocalState() != EndpointState.CLOSED) {
        link.close();
    }
}
 

import org.apache.qpid.proton.engine.EndpointState; //导入依赖的package包/类
@Override
public void onConnectionLocalOpen(Event evt) {
    Connection conn = evt.getConnection();
    if (conn.getRemoteState() == EndpointState.UNINITIALIZED) {
        // Give the connection a [random] container-id
        conn.setContainer(UUID.randomUUID().toString());
        try {
            new Connector(conn);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}
 

import org.apache.qpid.proton.engine.EndpointState; //导入依赖的package包/类
@Override
public Session next(EnumSet<EndpointState> local, EnumSet<EndpointState> remote)
{
    LinkNode.Query<SessionImpl> query = new EndpointImplQuery<SessionImpl>(local, remote);

    LinkNode<SessionImpl> sessionNode = _node.next(query);

    return sessionNode == null ? null : sessionNode.getValue();
}
 

import org.apache.qpid.proton.engine.EndpointState; //导入依赖的package包/类
/**
 * This method is public as it is used by Python layer.
 * @see org.apache.qpid.proton.engine.Transport#input(byte[], int, int)
 */
public TransportResult oldApiCheckStateBeforeInput(int inputLength)
{
    _lastTransportResult.checkIsOk();
    if(inputLength == 0)
    {
        if(_connectionEndpoint == null || _connectionEndpoint.getRemoteState() != EndpointState.CLOSED)
        {
            return TransportResultFactory.error(new TransportException("Unexpected EOS when remote connection not closed: connection aborted"));
        }
    }
    return TransportResultFactory.ok();
}