import org.apache.qpid.proton.engine.Session; //导入依赖的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.Session; //导入依赖的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.Session; //导入依赖的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.Session; //导入依赖的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.Session; //导入依赖的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.Session; //导入依赖的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.Session; //导入依赖的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.Session; //导入依赖的package包/类
@Override
public void onRemoteClose(Session session) throws Exception {
   lock();
   try {
      session.close();
      session.free();
   } finally {
      unlock();
   }

   AMQPSessionContext sessionContext = (AMQPSessionContext) session.getContext();
   if (sessionContext != null) {
      sessionContext.close();
      sessions.remove(session);
      session.setContext(null);
   }
}
 

import org.apache.qpid.proton.engine.Session; //导入依赖的package包/类
@Override
public void close() throws NetworkException, MessagingException, TimeoutException
{
    synchronized (_lock)
    {
        if (!_closed.get())
        {
            _closed.set(true);
            if (_connected.get())
            {
                for (Session ssn : _sessions.keySet())
                {
                    _sessions.get(ssn).closeImpl();
                }
                _sessions.clear();
                _connection.close();
                write();
                closeNetworkConnection();
            }
        }
    }
}
 

import org.apache.qpid.proton.engine.Session; //导入依赖的package包/类
private void createSender(org.apache.qpid.proton.engine.Session session) throws Exception {
   Sender sender = session.sender(subscriberInfo.getClientId());
   Target target = new Target();
   target.setAddress(subscriberInfo.getClientAddress());
   sender.setTarget(target);

   Source source = new Source();
   source.setAddress(subscriberInfo.getClientAddress());
   source.setDurable(TerminusDurability.UNSETTLED_STATE);
   sender.setSource(source);

   sender.open();
}
 

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

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

import org.apache.qpid.proton.engine.Session; //导入依赖的package包/类
public Sender newOutgoing(Session ssn, String remote, String local) {
    Sender snd = ssn.sender(String.format("%s-%s", local, remote));
    Source src = new Source();
    src.setAddress(local);
    snd.setSource(src);
    Target tgt = new Target();
    tgt.setAddress(remote);
    snd.setTarget(tgt);
    return snd;
}
 

import org.apache.qpid.proton.engine.Session; //导入依赖的package包/类
public Receiver newIncoming(Session ssn, String remote, String local) {
    Receiver rcv = ssn.receiver(String.format("%s-%s", remote, local));
    Source src = new Source();
    src.setAddress(remote);
    rcv.setSource(src);
    Target tgt = new Target();
    tgt.setAddress(remote);
    rcv.setTarget(tgt);
    return rcv;
}
 

import org.apache.qpid.proton.engine.Session; //导入依赖的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.Session; //导入依赖的package包/类
@Override
public Session sessionHead(final EnumSet<EndpointState> local, final EnumSet<EndpointState> remote)
{
    if(_sessionHead == null)
    {
        return null;
    }
    else
    {
        LinkNode.Query<SessionImpl> query = new EndpointImplQuery<SessionImpl>(local, remote);
        LinkNode<SessionImpl> node = query.matches(_sessionHead) ? _sessionHead : _sessionHead.next(query);
        return node == null ? null : node.getValue();
    }
}
 

import org.apache.qpid.proton.engine.Session; //导入依赖的package包/类
@Override
void doFree() {
    List<SessionImpl> sessions = new ArrayList<SessionImpl>(_sessions);
    for(Session session : sessions) {
        session.free();
    }
    _sessions = null;
}
 

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

import org.apache.qpid.proton.engine.Session; //导入依赖的package包/类
void doOpenLinkBeforeOpenConnectionTestImpl(boolean receiverLink)
{
    MockTransportImpl transport = new MockTransportImpl();
    Connection connection = Proton.connection();
    transport.bind(connection);

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

    Link link = null;
    if(receiverLink)
    {
        link = session.receiver("myReceiver");
    }
    else
    {
        link = session.sender("mySender");
    }
    link.open();

    pumpMockTransport(transport);

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

    // Now open the connection, expect the Open, Begin, and Attach frames
    connection.open();

    pumpMockTransport(transport);

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

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

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

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

    Receiver reciever = session.receiver("myReceiver");
    reciever.flow(5);

    pumpMockTransport(transport);

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

    // Now open the connection, expect the Open and Begin frames but
    // nothing else as we haven't opened the receiver itself yet.
    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);
}