private void processConnection(@Nonnull final HttpConnection connection, @Nonnull final AsyncContext context) throws HttpConnectionClosedException, IOException
    {
        final long rid = connection.getRequestId();
        final StreamID streamid = getStreamID();
        Log.debug( "Adding connection to stream {} with rid {}", streamid, rid );

        // Note that connections can be expected to arrive 'out of order'. The implementation should only use a connection
        // that has a request ID value that's exactly one higher than the last request ID value in the 'gap-less' sequence
        // of request IDs. When a connection is being processed that has a higher value, it should go unused until another
        // connection arrives that 'fills the gap' (and be used only _after_ that connection gets used).
        boolean aConnectionAvailableForDelivery = false;
        boolean mustClose = false;
        synchronized (connectionQueue) {
            // Note that this queue will automatically order its entities.
            connectionQueue.add(connection);

            lastActivity = Instant.now();

            final Iterator<HttpConnection> iter = connectionQueue.iterator();
            while (iter.hasNext()) {
                final HttpConnection queuedConnection = iter.next();
                assert !queuedConnection.isClosed();

                final long queuedRequestID = queuedConnection.getRequestId();
                if (queuedRequestID <= lastSequentialRequestID)
                {
                    // The request body (inbound data) for this request will already have been processed, but the connection remains queued, waiting to be used to send outbound data.
                    Log.trace("Detected a queued connection (for session {}) with a request ID ({}) that is not higher than the last sequential request ID ({}). This connection is waiting to be used to deliver outbound data back to the client.", streamid, queuedRequestID, lastSequentialRequestID);
                    continue;
                }
                if (queuedRequestID == lastSequentialRequestID + 1)
                {
                    Log.debug("Detected a queued connection (for session {}) with request ID ({}) that is exactly one higher than the last sequential request ID ({}). This inbound data on this connection will now be processed, after which the connection can be used to send outbound data back to the client.", streamid, queuedRequestID, lastSequentialRequestID);
                    // The data that was provided by the client can now be processed by the server.
                    // The below sends this data asynchronously.
                    sendPendingPackets(queuedConnection.getInboundDataQueue());

                    // Evaluate edge-cases.
                    if (queuedConnection.isTerminate()) {
                        Log.debug("Connection (for session {}) with request ID ({}) is a request to terminate.", getStreamID(), queuedRequestID);
                        iter.remove(); // This connection will be consumed here.
                        queuedConnection.deliverBody(createEmptyBody(true), true);
                        mustClose = true;
                    } else if (queuedConnection.isRestart()) {
                        Log.debug("Connection (for session {}) with request ID ({}) is a request to restart.", getStreamID(), queuedRequestID);
                        iter.remove(); // This connection has now been fully consumed.
                        queuedConnection.deliverBody(createSessionRestartResponse(), true);
                    } else if (queuedConnection.getPause() != null) {
                        // OF-2449: Error when the requested pause is higher than the allowed maximum.
                        if (!IGNORE_INVALID_PAUSE.getValue() && (queuedConnection.getPause().compareTo(getMaxPause()) > 0 || queuedConnection.getPause().isNegative())) {
                            Log.info("Connection (for session {}) with request ID ({}) is a request to pause (for {}) that is outside of the permissible range of 0 to {}", getStreamID(), queuedRequestID, queuedConnection.getPause(), getMaxPause());
                            queuedConnection.deliverBody(createTerminalBindingBody("policy-violation"), true);
                            mustClose = true;
                        } else {
                            Log.debug("Connection (for session {}) with request ID ({}) is a request to pause (for {}).", getStreamID(), queuedRequestID, queuedConnection.getPause());
                            pause(queuedConnection.getPause());
                            queuedConnection.deliverBody(createEmptyBody(false), true);
                            setLastResponseEmpty(true);
                        }
                        iter.remove(); // This connection will be consumed by this block. It should not be processed by the 'pause' method.
                    } else {
                        // At least one new connection has become available, and can be used to return data to the client.
                        aConnectionAvailableForDelivery = true;
                    }

                    // There is a new connection available for consumption.
                    lastSequentialRequestID = queuedRequestID;

                    // Note that when this connection fills a gap in the sequence, other connections might already be
                    // available that have the 'next' Request ID value. We need to keep iterating.
                } else {
                    // As we're iterating over the collection that is ordered by Request ID, the iteration can stop
                    // when a gap is detected (subsequent connections will only have higher Request ID values).
                    break;
                }
            }

            if (!mustClose) {
                // If a connection became available for delivery and there's pending data to be delivered, deliver immediately.
                // Request ID of the new connection 'fits in the window'

                if (isPollingSession()) {
                    // Note that the code leading up to here checks if the Request ID of the new connection 'fits in the window',
                    // which means that for polling sessions, the request ID must have been a sequential one, which in turn should
                    // guarantee that 'a new connection is now available for delivery').
                    assert aConnectionAvailableForDelivery; // FIXME: OF-2451: the edge-cases evaluated above make this assertion not necessarily true.
                }

                if (isPollingSession() || aConnectionAvailableForDelivery) {
                    SessionEventDispatcher.dispatchEvent(this, SessionEventDispatcher.EventType.connection_opened, connection, context); // TODO is this the right place to dispatch this event?
                    tryImmediateDelivery();
                }

                // When a new connection has become available, older connections need to be released (allowing the client to
                // send more data if it needs to).
                while (!connectionQueue.isEmpty() && connectionQueue.size() > hold) {
                    Log.trace("Stream {}: releasing oldest connection (rid {}), as the amount of open connections ({}) is higher than the requested amount to hold ({}).", streamid, rid, connectionQueue.size(), hold);
                    final HttpConnection openConnection = connectionQueue.peek();
                    assert openConnection != null;
                    if (openConnection.getRequestId() > lastSequentialRequestID) {
                        break; // There's a gap. As described above, connections must be used in sequence, without jumping the queue.
                    }

                    // Consume this connection.
                    connectionQueue.poll();
                    openConnection.deliverBody(createEmptyBody(false), true);
                }
            }
        }

        // OF-2444: Call 'close()' outside of the connectionQueue mutex, to avoid deadlocks.
        if (mustClose) {
            close();
        }
    }
Schedules the connection for processing, ensuring that connections are processed in the order of their request ID
values. Processing causes data delivered by the client to be routed in the server to their intended recipients,
and allow the connection to be used to send back data to the client.

@param connection The connection ready to be processed.
@param context the context of the asynchronous servlet call leading up to this method call.
xmppserver/src/main/java/org/jivesoftware/openfire/http/HttpSession.java