import org.apache.qpid.proton.engine.Connection; //导入依赖的package包/类
private void getUserAndRolesFromConnection(Connection connection) {
    final Map<Symbol, Object> remoteProperties = connection.getRemoteProperties();
    if (remoteProperties != null && remoteProperties.get(AUTHENTICATED_IDENTITY) instanceof Map) {
        Map identity = (Map) remoteProperties.get(AUTHENTICATED_IDENTITY);
        if (identity.containsKey(PREFERRED_USERNAME)) {
            user = String.valueOf(identity.get(PREFERRED_USERNAME)).trim();
        } else {
            user = String.valueOf(identity.get(SUB)).trim();
        }
        if (remoteProperties.get(Symbol.valueOf(GROUPS)) instanceof List) {
            List<String> groups = (List<String>) remoteProperties.get(Symbol.valueOf(GROUPS));
            roles.addAll(groups);
            if(this.securitySettings != null) {
                SaslGroupBasedSecuritySettingsPlugin securitySettingPlugin = SaslGroupBasedSecuritySettingsPlugin.getInstance(this.securitySettings);
                if(securitySettingPlugin != null) {
                    securitySettingPlugin.addGroups(groups);
                }
            }
        }
        roles.add("all");

    }
}
 

import org.apache.qpid.proton.engine.Connection; //导入依赖的package包/类
private void performSaslSteps(Connection connection, InputStream in,
                              OutputStream out,
                              SaslMechanism mechanism) throws IOException, LoginException {
    Transport transport = connection.getTransport();
    Sasl sasl = transport.sasl();
    do {

        readFromNetwork(connection, in, () ->
            !(EnumSet.of(PN_SASL_PASS, PN_SASL_FAIL).contains(sasl.getState())
            || (sasl.getState() == PN_SASL_STEP && sasl.pending() > 0)));

        if (sasl.pending() > 0) {
            byte[] challenge = new byte[sasl.pending()];
            byte[] response = mechanism.getResponse(challenge);
            if (sasl.getState() == PN_SASL_STEP) {
                sasl.send(response, 0, response.length);
                writeToNetwork(connection, out);
            }
        }

    } while (sasl.getState() == PN_SASL_STEP);
}
 

import org.apache.qpid.proton.engine.Connection; //导入依赖的package包/类
private void readFromNetwork(Connection connection, InputStream in, BooleanSupplier test) throws IOException, LoginException {
    Transport transport = connection.getTransport();
    while(test.getAsBoolean()) {
        ByteBuffer buf = transport.getInputBuffer();
        byte[] tmpBuf = new byte[buf.remaining()];
        int bytesRead = in.read(tmpBuf);
        LOG.tracev("read {0} bytes", bytesRead);
        if (bytesRead == -1) {
            throw new LoginException("Unexpected EOS experienced when authenticating using SASL delegation");
        } else {
            buf.put(tmpBuf, 0, bytesRead);
            TransportResult result = transport.processInput();
            if(!result.isOk()) {
                LoginException e = new LoginException("Unexpected error when authenticating using SASL delegation");
                e.initCause(result.getException());
                throw e;
            }
        }

    }

}
 

import org.apache.qpid.proton.engine.Connection; //导入依赖的package包/类
public Pool(Collector collector, final Router router) {
    this.collector = collector;
    connections = new HashMap<String,Connection>();

    if (router != null) {
        outgoingResolver = new LinkResolver<Sender>() {
            public Sender resolve(String address) {
                return router.getOutgoing(address).choose();
            }
        };
        incomingResolver = new LinkResolver<Receiver>() {
            public Receiver resolve(String address) {
                return router.getIncoming(address).choose();
            }
        };
    } else {
        outgoingResolver = new LinkResolver<Sender>() {
            public Sender resolve(String address) { return null; }
        };
        incomingResolver = new LinkResolver<Receiver>() {
            public Receiver resolve(String address) { return null; }
        };
    }
}
 

import org.apache.qpid.proton.engine.Connection; //导入依赖的package包/类
private <T extends Link> T resolve(String remote, String local,
                                   LinkResolver<T> resolver,
                                   LinkConstructor<T> constructor) {
    String host = remote.substring(2).split("/", 2)[0];
    T link = resolver.resolve(remote);
    if (link == null) {
        Connection conn = connections.get(host);
        if (conn == null) {
            conn = Connection.Factory.create();
            conn.collect(collector);
            conn.setHostname(host);
            conn.open();
            connections.put(host, conn);
        }

        Session ssn = conn.session();
        ssn.open();

        link = constructor.create(ssn, remote, local);
        link.open();
    }
    return link;
}
 

import org.apache.qpid.proton.engine.Connection; //导入依赖的package包/类
@Override
public void onConnectionInit(Event event) {
    Connection conn = event.getConnection();

    // Every session or link could have their own handler(s) if we
    // wanted simply by adding the handler to the given session
    // or link
    Session ssn = conn.session();

    // If a link doesn't have an event handler, the events go to
    // its parent session. If the session doesn't have a handler
    // the events go to its parent connection. If the connection
    // doesn't have a handler, the events go to the reactor.
    Sender snd = ssn.sender("sender");
    conn.open();
    ssn.open();
    snd.open();
}
 

import org.apache.qpid.proton.engine.Connection; //导入依赖的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.Connection; //导入依赖的package包/类
@Override
public Connection getConnection()
{
    if (context instanceof Connection) {
        return (Connection) context;
    } else if (context instanceof Transport) {
        Transport transport = getTransport();
        if (transport == null) {
            return null;
        }
        return ((TransportImpl) transport).getConnectionImpl();
    } else {
        Session ssn = getSession();
        if (ssn == null) {
            return null;
        }
        return ssn.getConnection();
    }
}
 

import org.apache.qpid.proton.engine.Connection; //导入依赖的package包/类
@Override
public Transport getTransport()
{
    if (context instanceof Transport) {
        return (Transport) context;
    } else if (context instanceof Connection) {
        return ((Connection)context).getTransport();
    } else {
        Session session = getSession();
        if (session == null) {
            return null;
        }

        Connection connection = session.getConnection();
        if (connection == null) {
            return null;
        }

        return connection.getTransport();
    }
}
 

import org.apache.qpid.proton.engine.Connection; //导入依赖的package包/类
@Override
public Reactor getReactor() {
    if (context instanceof Reactor) {
        return (Reactor) context;
    } else if (context instanceof Task) {
        return ((Task)context).getReactor();
    } else if (context instanceof Transport) {
        return ((TransportImpl)context).getReactor();
    } else if (context instanceof Delivery) {
        return ((Delivery)context).getLink().getSession().getConnection().getReactor();
    } else if (context instanceof Link) {
        return ((Link)context).getSession().getConnection().getReactor();
    } else if (context instanceof Session) {
        return ((Session)context).getConnection().getReactor();
    } else if (context instanceof Connection) {
        return ((Connection)context).getReactor();
    } else if (context instanceof Selectable) {
        return ((Selectable)context).getReactor();
    }
    return null;
}
 

import org.apache.qpid.proton.engine.Connection; //导入依赖的package包/类
@Override
public void setConnectionHost(Connection connection,
                              String host, int port) {
    Record r = connection.attachments();
    // cannot set the address on an incoming connection
    if (r.get(AcceptorImpl.CONNECTION_ACCEPTOR_KEY, Acceptor.class) == null) {
        Address addr = new Address();
        addr.setHost(host);
        if (port == 0) {
            port = 5672;
        }
        addr.setPort(Integer.toString(port));
        r.set(CONNECTION_PEER_ADDRESS_KEY, Address.class, addr);
    } else {
        throw new IllegalStateException("Cannot set the host address on an incoming Connection");
    }
}
 

import org.apache.qpid.proton.engine.Connection; //导入依赖的package包/类
@Test
public void testOutputTooBigToBeWrittenInOneGo()
{
    int smallMaxFrameSize = 512;
    _transport = new TransportImpl(smallMaxFrameSize);

    Connection conn = new ConnectionImpl();
    _transport.bind(conn);

    // Open frame sized in order to produce a frame that will almost fill output buffer
    conn.setHostname(stringOfLength("x", 500));
    conn.open();

    // Close the connection to generate a Close frame which will cause an overflow
    // internally - we'll get the remaining bytes on the next interaction.
    conn.close();

    ByteBuffer buf = _transport.getOutputBuffer();
    assertEquals("Expecting buffer to be full", smallMaxFrameSize, buf.remaining());
    buf.position(buf.limit());
    _transport.outputConsumed();

    buf  = _transport.getOutputBuffer();
    assertTrue("Expecting second buffer to have bytes", buf.remaining() > 0);
    assertTrue("Expecting second buffer to not be full", buf.remaining() < Transport.MIN_MAX_FRAME_SIZE);
}
 

import org.apache.qpid.proton.engine.Connection; //导入依赖的package包/类
@Test
public void testOpenSessionBeforeOpenConnection()
{
    MockTransportImpl transport = new MockTransportImpl();
    Connection connection = Proton.connection();
    transport.bind(connection);

    Session session = connection.session();
    session.open();

    pumpMockTransport(transport);

    assertEquals("Unexpected frames written: " + getFrameTypesWritten(transport), 0, transport.writes.size());

    connection.open();

    pumpMockTransport(transport);

    assertEquals("Unexpected frames written: " + getFrameTypesWritten(transport), 2, transport.writes.size());

    assertTrue("Unexpected frame type", transport.writes.get(0) instanceof Open);
    assertTrue("Unexpected frame type", transport.writes.get(1) instanceof Begin);
}
 

import org.apache.qpid.proton.engine.Connection; //导入依赖的package包/类
@Test
public void testGetTransportWithDeliveryContext()
{
    Transport transport = Transport.Factory.create();
    Connection connection = Connection.Factory.create();
    transport.bind(connection);

    Session session = connection.session();
    Sender sender = session.sender("mySender");

    Delivery delivery = sender.delivery("tag".getBytes());

    EventImpl event = createEvent(delivery, Event.Type.DELIVERY);

    assertNotNull("No transport returned", event.getTransport());
    assertSame("Incorrect transport returned", transport, event.getTransport());
}
 

import org.apache.qpid.proton.engine.Connection; //导入依赖的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.Connection; //导入依赖的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.Connection; //导入依赖的package包/类
@Override
public void onAuthInit(ProtonHandler handler, Connection connection, boolean sasl) {
   if (sasl) {
      // configured mech in decreasing order of preference
      String[] mechanisms = connectionCallback.getSaslMechanisms();
      if (mechanisms == null || mechanisms.length == 0) {
         mechanisms = AnonymousServerSASL.ANONYMOUS_MECH;
      }
      handler.createServerSASL(mechanisms);
   } else {
      if (!connectionCallback.isSupportsAnonymous()) {
         connectionCallback.sendSASLSupported();
         connectionCallback.close();
         handler.close(null);
      }
   }
}
 

import org.apache.qpid.proton.engine.Connection; //导入依赖的package包/类
@Override
public void onRemoteClose(Connection connection) {
   lock();
   try {
      connection.close();
      connection.free();
   } finally {
      unlock();
   }

   for (AMQPSessionContext protonSession : sessions.values()) {
      protonSession.close();
   }
   sessions.clear();

   // We must force write the channel before we actually destroy the connection
   handler.flushBytes();
   destroy();
}
 

import org.apache.qpid.proton.engine.Connection; //导入依赖的package包/类
ProtonTransport(Connection connection, Vertx vertx, NetClient netClient, NetSocket socket,
                ProtonSaslAuthenticator authenticator, ProtonTransportOptions options) {
  this.connection = connection;
  this.vertx = vertx;
  this.netClient = netClient;
  this.socket = socket;
  transport.setMaxFrameSize(options.getMaxFrameSize() == 0 ? DEFAULT_MAX_FRAME_SIZE : options.getMaxFrameSize());
  transport.setEmitFlowEventOnSend(false); // TODO: make configurable
  transport.setIdleTimeout(2 * options.getHeartbeat());
  if (authenticator != null) {
    authenticator.init(this.socket, (ProtonConnection) this.connection.getContext(), transport);
  }
  this.authenticator = authenticator;
  transport.bind(connection);
  connection.collect(collector);
  socket.endHandler(this::handleSocketEnd);
  socket.handler(this::handleSocketBuffer);
}
 

import org.apache.qpid.proton.engine.Connection; //导入依赖的package包/类
@Override
public void channelActive(ChannelHandlerContext ctx) {
    synchronized (lock) {
        System.out.println("ACTIVE");
        transport = Transport.Factory.create();
        transport.setContext(ctx);

        Sasl sasl = transport.sasl();
        sasl.setMechanisms("ANONYMOUS");
        sasl.server();
        sasl.done(Sasl.PN_SASL_OK);

        connection = Connection.Factory.create();
        collector = Collector.Factory.create();
        connection.collect(collector);
        transport.bind(connection);

        // XXX: really we should fire both of these off of an
        //      initial transport event
        write(ctx);
        int capacity = transport.capacity();
        if (capacity > 0) {
            ctx.read();
        }
    }
}