| /* |
| * |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| * |
| */ |
| package org.apache.qpid.server.protocol.v1_0; |
| |
| import static com.google.common.util.concurrent.Futures.allAsList; |
| import static org.apache.qpid.server.protocol.v1_0.type.extensions.soleconn.SoleConnectionConnectionProperties.SOLE_CONNECTION_ENFORCEMENT_POLICY; |
| |
| import java.net.SocketAddress; |
| import java.security.AccessControlContext; |
| import java.security.AccessControlException; |
| import java.security.AccessController; |
| import java.security.Principal; |
| import java.security.PrivilegedAction; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.Iterator; |
| import java.util.LinkedHashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.NoSuchElementException; |
| import java.util.Optional; |
| import java.util.Queue; |
| import java.util.Set; |
| import java.util.UUID; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.ConcurrentLinkedQueue; |
| import java.util.concurrent.ConcurrentSkipListMap; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| import java.util.concurrent.atomic.AtomicReference; |
| import java.util.stream.Collectors; |
| import java.util.stream.StreamSupport; |
| |
| import com.google.common.collect.Iterators; |
| import com.google.common.collect.PeekingIterator; |
| import com.google.common.collect.Sets; |
| import com.google.common.util.concurrent.ListenableFuture; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import org.apache.qpid.server.bytebuffer.QpidByteBuffer; |
| import org.apache.qpid.server.logging.messages.ConnectionMessages; |
| import org.apache.qpid.server.model.AuthenticationProvider; |
| import org.apache.qpid.server.model.Broker; |
| import org.apache.qpid.server.model.Connection; |
| import org.apache.qpid.server.model.NamedAddressSpace; |
| import org.apache.qpid.server.model.Protocol; |
| import org.apache.qpid.server.model.Transport; |
| import org.apache.qpid.server.model.port.AmqpPort; |
| import org.apache.qpid.server.plugin.ConnectionPropertyEnricher; |
| import org.apache.qpid.server.protocol.ConnectionClosingTicker; |
| import org.apache.qpid.server.protocol.v1_0.codec.DescribedTypeConstructorRegistry; |
| import org.apache.qpid.server.protocol.v1_0.codec.FrameWriter; |
| import org.apache.qpid.server.protocol.v1_0.codec.ProtocolHandler; |
| import org.apache.qpid.server.protocol.v1_0.codec.SectionDecoderRegistry; |
| import org.apache.qpid.server.protocol.v1_0.codec.ValueHandler; |
| import org.apache.qpid.server.protocol.v1_0.codec.ValueWriter; |
| import org.apache.qpid.server.protocol.v1_0.framing.AMQFrame; |
| import org.apache.qpid.server.protocol.v1_0.framing.FrameHandler; |
| import org.apache.qpid.server.protocol.v1_0.framing.OversizeFrameException; |
| import org.apache.qpid.server.protocol.v1_0.framing.SASLFrame; |
| import org.apache.qpid.server.protocol.v1_0.framing.TransportFrame; |
| import org.apache.qpid.server.protocol.v1_0.type.Binary; |
| import org.apache.qpid.server.protocol.v1_0.type.ErrorCondition; |
| import org.apache.qpid.server.protocol.v1_0.type.FrameBody; |
| import org.apache.qpid.server.protocol.v1_0.type.SaslFrameBody; |
| import org.apache.qpid.server.protocol.v1_0.type.Symbol; |
| import org.apache.qpid.server.protocol.v1_0.type.UnsignedInteger; |
| import org.apache.qpid.server.protocol.v1_0.type.UnsignedShort; |
| import org.apache.qpid.server.protocol.v1_0.type.codec.AMQPDescribedTypeRegistry; |
| import org.apache.qpid.server.protocol.v1_0.type.extensions.soleconn.SoleConnectionConnectionProperties; |
| import org.apache.qpid.server.protocol.v1_0.type.extensions.soleconn.SoleConnectionDetectionPolicy; |
| import org.apache.qpid.server.protocol.v1_0.type.extensions.soleconn.SoleConnectionEnforcementPolicy; |
| import org.apache.qpid.server.protocol.v1_0.type.security.SaslChallenge; |
| import org.apache.qpid.server.protocol.v1_0.type.security.SaslCode; |
| import org.apache.qpid.server.protocol.v1_0.type.security.SaslInit; |
| import org.apache.qpid.server.protocol.v1_0.type.security.SaslMechanisms; |
| import org.apache.qpid.server.protocol.v1_0.type.security.SaslOutcome; |
| import org.apache.qpid.server.protocol.v1_0.type.security.SaslResponse; |
| import org.apache.qpid.server.protocol.v1_0.type.transport.AmqpError; |
| import org.apache.qpid.server.protocol.v1_0.type.transport.Attach; |
| import org.apache.qpid.server.protocol.v1_0.type.transport.Begin; |
| import org.apache.qpid.server.protocol.v1_0.type.transport.ChannelFrameBody; |
| import org.apache.qpid.server.protocol.v1_0.type.transport.Close; |
| import org.apache.qpid.server.protocol.v1_0.type.transport.ConnectionError; |
| import org.apache.qpid.server.protocol.v1_0.type.transport.Detach; |
| import org.apache.qpid.server.protocol.v1_0.type.transport.Disposition; |
| import org.apache.qpid.server.protocol.v1_0.type.transport.End; |
| import org.apache.qpid.server.protocol.v1_0.type.transport.Error; |
| import org.apache.qpid.server.protocol.v1_0.type.transport.Flow; |
| import org.apache.qpid.server.protocol.v1_0.type.transport.Open; |
| import org.apache.qpid.server.protocol.v1_0.type.transport.Transfer; |
| import org.apache.qpid.server.security.SubjectCreator; |
| import org.apache.qpid.server.security.auth.AuthenticatedPrincipal; |
| import org.apache.qpid.server.security.auth.AuthenticationResult; |
| import org.apache.qpid.server.security.auth.SubjectAuthenticationResult; |
| import org.apache.qpid.server.security.auth.manager.AnonymousAuthenticationManager; |
| import org.apache.qpid.server.security.auth.manager.ExternalAuthenticationManagerImpl; |
| import org.apache.qpid.server.security.auth.sasl.SaslNegotiator; |
| import org.apache.qpid.server.session.AMQPSession; |
| import org.apache.qpid.server.transport.AMQPConnection; |
| import org.apache.qpid.server.transport.AbstractAMQPConnection; |
| import org.apache.qpid.server.transport.AggregateTicker; |
| import org.apache.qpid.server.transport.ByteBufferSender; |
| import org.apache.qpid.server.transport.ProtocolEngine; |
| import org.apache.qpid.server.transport.ServerNetworkConnection; |
| import org.apache.qpid.server.transport.util.Functions; |
| import org.apache.qpid.server.txn.LocalTransaction; |
| import org.apache.qpid.server.txn.ServerTransaction; |
| import org.apache.qpid.server.util.Action; |
| import org.apache.qpid.server.util.ConnectionScopedRuntimeException; |
| import org.apache.qpid.server.util.ServerScopedRuntimeException; |
| import org.apache.qpid.server.virtualhost.VirtualHostUnavailableException; |
| |
| public class AMQPConnection_1_0Impl extends AbstractAMQPConnection<AMQPConnection_1_0Impl, ConnectionHandler> |
| implements DescribedTypeConstructorRegistry.Source, |
| ValueWriter.Registry.Source, |
| SASLEndpoint, |
| AMQPConnection_1_0<AMQPConnection_1_0Impl> |
| { |
| |
| private static final Logger LOGGER = LoggerFactory.getLogger(AMQPConnection_1_0Impl.class); |
| private static final Logger FRAME_LOGGER = LoggerFactory.getLogger("org.apache.qpid.server.protocol.frame"); |
| |
| private final AtomicBoolean _stateChanged = new AtomicBoolean(); |
| private final AtomicReference<Action<ProtocolEngine>> _workListener = new AtomicReference<>(); |
| |
| |
| private static final byte[] SASL_HEADER = new byte[] |
| { |
| (byte) 'A', |
| (byte) 'M', |
| (byte) 'Q', |
| (byte) 'P', |
| (byte) 3, |
| (byte) 1, |
| (byte) 0, |
| (byte) 0 |
| }; |
| |
| private static final byte[] AMQP_HEADER = new byte[] |
| { |
| (byte) 'A', |
| (byte) 'M', |
| (byte) 'Q', |
| (byte) 'P', |
| (byte) 0, |
| (byte) 1, |
| (byte) 0, |
| (byte) 0 |
| }; |
| |
| private final FrameWriter _frameWriter; |
| private ProtocolHandler _frameHandler; |
| private volatile boolean _transportBlockedForWriting; |
| private volatile SubjectAuthenticationResult _successfulAuthenticationResult; |
| private boolean _blocking; |
| private final Object _blockingLock = new Object(); |
| private List<Symbol> _offeredCapabilities; |
| private SoleConnectionEnforcementPolicy _soleConnectionEnforcementPolicy; |
| |
| private static final int CONNECTION_CONTROL_CHANNEL = 0; |
| private final SubjectCreator _subjectCreator; |
| |
| private int _channelMax = 0; |
| private int _maxFrameSize = 4096; |
| private String _remoteContainerId; |
| |
| private SocketAddress _remoteAddress; |
| |
| // positioned by the *outgoing* channel |
| private Session_1_0[] _sendingSessions; |
| |
| // positioned by the *incoming* channel |
| private Session_1_0[] _receivingSessions; |
| private volatile boolean _closedForOutput; |
| |
| private final long _incomingIdleTimeout; |
| |
| private volatile long _outgoingIdleTimeout; |
| |
| private volatile ConnectionState _connectionState = ConnectionState.AWAIT_AMQP_OR_SASL_HEADER; |
| |
| private final AMQPDescribedTypeRegistry _describedTypeRegistry = AMQPDescribedTypeRegistry.newInstance() |
| .registerTransportLayer() |
| .registerMessagingLayer() |
| .registerTransactionLayer() |
| .registerSecurityLayer() |
| .registerExtensionSoleconnLayer(); |
| |
| private final Map<Symbol, Object> _properties = new LinkedHashMap<>(); |
| private volatile boolean _saslComplete; |
| |
| private volatile SaslNegotiator _saslNegotiator; |
| private String _localHostname; |
| |
| private static final long MINIMUM_SUPPORTED_IDLE_TIMEOUT = 1000L; |
| |
| private Set<Symbol> _remoteDesiredCapabilities; |
| |
| private final AtomicBoolean _orderlyClose = new AtomicBoolean(false); |
| |
| private final Collection<Session_1_0> |
| _sessions = Collections.synchronizedCollection(new ArrayList<Session_1_0>()); |
| |
| private final Object _reference = new Object(); |
| |
| private final Queue<Action<? super ConnectionHandler>> _asyncTaskList = |
| new ConcurrentLinkedQueue<>(); |
| |
| private final Set<AMQPSession<?,?>> _sessionsWithWork = |
| Collections.newSetFromMap(new ConcurrentHashMap<AMQPSession<?,?>, Boolean>()); |
| |
| |
| // Multi session transactions |
| private final Map<Integer, ServerTransaction> _openTransactions = new ConcurrentSkipListMap<>(); |
| private volatile boolean _sendSaslFinalChallengeAsChallenge; |
| private volatile String _closeCause; |
| |
| AMQPConnection_1_0Impl(final Broker<?> broker, |
| final ServerNetworkConnection network, |
| AmqpPort<?> port, |
| Transport transport, |
| long id, |
| final AggregateTicker aggregateTicker) |
| { |
| super(broker, network, port, transport, Protocol.AMQP_1_0, id, aggregateTicker); |
| |
| _subjectCreator = port.getSubjectCreator(transport.isSecure(), network.getSelectedHost()); |
| |
| List<Symbol> offeredCapabilities = new ArrayList<>(); |
| offeredCapabilities.add(ANONYMOUS_RELAY); |
| offeredCapabilities.add(SHARED_SUBSCRIPTIONS); |
| offeredCapabilities.add(SoleConnectionConnectionProperties.SOLE_CONNECTION_FOR_CONTAINER); |
| |
| setOfferedCapabilities(offeredCapabilities); |
| |
| setRemoteAddress(network.getRemoteAddress()); |
| |
| _incomingIdleTimeout = 1000L * port.getHeartbeatDelay(); |
| |
| _frameWriter = new FrameWriter(getDescribedTypeRegistry(), getSender()); |
| } |
| |
| @Override |
| protected void onOpen() |
| { |
| super.onOpen(); |
| _sendSaslFinalChallengeAsChallenge = getContextValue(Boolean.class, AMQPConnection_1_0.SEND_SASL_FINAL_CHALLENGE_AS_CHALLENGE); |
| } |
| |
| @Override |
| public void receiveSaslInit(final SaslInit saslInit) |
| { |
| assertState(ConnectionState.AWAIT_SASL_INIT); |
| if(saslInit.getHostname() != null && !"".equals(saslInit.getHostname().trim())) |
| { |
| _localHostname = saslInit.getHostname(); |
| } |
| else if(getNetwork().getSelectedHost() != null) |
| { |
| _localHostname = getNetwork().getSelectedHost(); |
| } |
| String mechanism = saslInit.getMechanism().toString(); |
| final Binary initialResponse = saslInit.getInitialResponse(); |
| byte[] response = initialResponse == null ? new byte[0] : initialResponse.getArray(); |
| |
| List<String> availableMechanisms = |
| _subjectCreator.getAuthenticationProvider().getAvailableMechanisms(getTransport().isSecure()); |
| if (!availableMechanisms.contains(mechanism)) |
| { |
| handleSaslError(); |
| } |
| else |
| { |
| _saslNegotiator = _subjectCreator.createSaslNegotiator(mechanism, this); |
| processSaslResponse(response); |
| } |
| } |
| |
| @Override |
| public void receiveSaslResponse(final SaslResponse saslResponse) |
| { |
| assertState(ConnectionState.AWAIT_SASL_RESPONSE); |
| final Binary responseBinary = saslResponse.getResponse(); |
| byte[] response = responseBinary == null ? new byte[0] : responseBinary.getArray(); |
| |
| processSaslResponse(response); |
| } |
| |
| @Override |
| public void receiveSaslMechanisms(final SaslMechanisms saslMechanisms) |
| { |
| LOGGER.info("{} : Unexpected frame sasl-mechanisms", getLogSubject()); |
| closeSaslWithFailure(); |
| } |
| |
| @Override |
| public void receiveSaslChallenge(final SaslChallenge saslChallenge) |
| { |
| LOGGER.info("{} : Unexpected frame sasl-challenge", getLogSubject()); |
| closeSaslWithFailure(); |
| } |
| |
| @Override |
| public void receiveSaslOutcome(final SaslOutcome saslOutcome) |
| { |
| LOGGER.info("{} : Unexpected frame sasl-outcome", getLogSubject()); |
| closeSaslWithFailure(); |
| } |
| |
| private void processSaslResponse(final byte[] response) |
| { |
| byte[] challenge = null; |
| SubjectAuthenticationResult authenticationResult = _successfulAuthenticationResult; |
| if (authenticationResult == null) |
| { |
| authenticationResult = _subjectCreator.authenticate(_saslNegotiator, response != null ? response : new byte[0]); |
| challenge = authenticationResult.getChallenge(); |
| } |
| |
| if (authenticationResult.getStatus() == AuthenticationResult.AuthenticationStatus.SUCCESS) |
| { |
| final boolean finalChallenge = challenge != null && challenge.length != 0; |
| _successfulAuthenticationResult = authenticationResult; |
| if (_sendSaslFinalChallengeAsChallenge && finalChallenge) |
| { |
| continueSaslNegotiation(challenge); |
| } |
| else |
| { |
| setSubject(_successfulAuthenticationResult.getSubject()); |
| SaslOutcome outcome = new SaslOutcome(); |
| outcome.setCode(SaslCode.OK); |
| if (finalChallenge) |
| { |
| outcome.setAdditionalData(new Binary(challenge)); |
| } |
| send(new SASLFrame(outcome)); |
| _saslComplete = true; |
| _connectionState = ConnectionState.AWAIT_AMQP_HEADER; |
| disposeSaslNegotiator(); |
| } |
| } |
| else if(authenticationResult.getStatus() == AuthenticationResult.AuthenticationStatus.CONTINUE) |
| { |
| continueSaslNegotiation(challenge); |
| } |
| else |
| { |
| handleSaslError(); |
| } |
| } |
| |
| private void continueSaslNegotiation(final byte[] challenge) |
| { |
| SaslChallenge challengeBody = new SaslChallenge(); |
| challengeBody.setChallenge(new Binary(challenge)); |
| send(new SASLFrame(challengeBody)); |
| |
| _connectionState = ConnectionState.AWAIT_SASL_RESPONSE; |
| } |
| |
| private void handleSaslError() |
| { |
| SaslOutcome outcome = new SaslOutcome(); |
| outcome.setCode(SaslCode.AUTH); |
| send(new SASLFrame(outcome)); |
| _saslComplete = true; |
| closeSaslWithFailure(); |
| } |
| |
| private void closeSaslWithFailure() |
| { |
| _saslComplete = true; |
| disposeSaslNegotiator(); |
| _connectionState = ConnectionState.CLOSED; |
| addCloseTicker(); |
| } |
| |
| private void disposeSaslNegotiator() |
| { |
| if (_saslNegotiator != null) |
| { |
| _saslNegotiator.dispose(); |
| } |
| _saslNegotiator = null; |
| } |
| |
| private void setUserPrincipal(final Principal user) |
| { |
| setSubject(_subjectCreator.createSubjectWithGroups(user)); |
| } |
| |
| @Override |
| public long getIncomingIdleTimeout() |
| { |
| return _incomingIdleTimeout; |
| } |
| |
| @Override |
| public long getOutgoingIdleTimeout() |
| { |
| return _outgoingIdleTimeout; |
| } |
| |
| |
| @Override |
| public void receiveAttach(final int channel, final Attach attach) |
| { |
| assertState(ConnectionState.OPENED); |
| final Session_1_0 session = getSession(channel); |
| if (session != null) |
| { |
| session.receiveAttach(attach); |
| } |
| else |
| { |
| closeConnection(AmqpError.INVALID_FIELD, "Channel " + channel + " is not associated with a session"); |
| } |
| } |
| |
| @Override |
| public void receive(final List<ChannelFrameBody> channelFrameBodies) |
| { |
| if (!channelFrameBodies.isEmpty()) |
| { |
| PeekingIterator<ChannelFrameBody> itr = Iterators.peekingIterator(channelFrameBodies.iterator()); |
| boolean cleanExit = false; |
| try |
| { |
| while (itr.hasNext()) |
| { |
| final ChannelFrameBody channelFrameBody = itr.next(); |
| final int frameChannel = channelFrameBody.getChannel(); |
| |
| Session_1_0 session = _receivingSessions == null || frameChannel >= _receivingSessions.length |
| ? null |
| : _receivingSessions[frameChannel]; |
| if (session != null) |
| { |
| final AccessControlContext context = session.getAccessControllerContext(); |
| AccessController.doPrivileged((PrivilegedAction<Void>) () -> |
| { |
| ChannelFrameBody channelFrame = channelFrameBody; |
| boolean nextIsSameChannel; |
| do |
| { |
| received(frameChannel, channelFrame.getFrameBody()); |
| nextIsSameChannel = itr.hasNext() && frameChannel == itr.peek().getChannel(); |
| if (nextIsSameChannel) |
| { |
| channelFrame = itr.next(); |
| } |
| } |
| while (nextIsSameChannel); |
| return null; |
| }, context); |
| } |
| else |
| { |
| received(frameChannel, channelFrameBody.getFrameBody()); |
| } |
| } |
| cleanExit = true; |
| } |
| finally |
| { |
| if (!cleanExit) |
| { |
| while (itr.hasNext()) |
| { |
| final Object frameBody = itr.next().getFrameBody(); |
| if (frameBody instanceof Transfer) |
| { |
| ((Transfer) frameBody).dispose(); |
| } |
| } |
| } |
| } |
| } |
| } |
| |
| private void received(int channel, Object val) |
| { |
| if (channel > getChannelMax()) |
| { |
| Error error = new Error(ConnectionError.FRAMING_ERROR, |
| String.format("specified channel %d larger than maximum channel %d", channel, getChannelMax())); |
| handleError(error); |
| return; |
| } |
| |
| FRAME_LOGGER.debug("RECV[{}|{}] : {}", _remoteAddress, channel, val); |
| if (val instanceof FrameBody) |
| { |
| ((FrameBody) val).invoke(channel, this); |
| } |
| else if (val instanceof SaslFrameBody) |
| { |
| ((SaslFrameBody) val).invoke(channel, this); |
| } |
| } |
| |
| |
| @Override |
| public void receiveClose(final int channel, final Close close) |
| { |
| switch (_connectionState) |
| { |
| case AWAIT_AMQP_OR_SASL_HEADER: |
| case AWAIT_SASL_INIT: |
| case AWAIT_SASL_RESPONSE: |
| case AWAIT_AMQP_HEADER: |
| throw new ConnectionScopedRuntimeException("Received unexpected close when AMQP connection has not been established."); |
| case AWAIT_OPEN: |
| closeReceived(); |
| closeConnection(ConnectionError.CONNECTION_FORCED, |
| "Connection close sent before connection was opened"); |
| break; |
| case OPENED: |
| _connectionState = ConnectionState.CLOSE_RECEIVED; |
| closeReceived(); |
| if(close.getError() != null) |
| { |
| final Error error = close.getError(); |
| ErrorCondition condition = error.getCondition(); |
| Symbol errorCondition = condition == null ? null : condition.getValue(); |
| LOGGER.info("{} : Connection closed with error : {} - {}", getLogSubject(), |
| errorCondition, close.getError().getDescription()); |
| } |
| sendClose(new Close()); |
| _connectionState = ConnectionState.CLOSED; |
| _orderlyClose.set(true); |
| addCloseTicker(); |
| break; |
| case CLOSE_SENT: |
| closeReceived(); |
| _connectionState = ConnectionState.CLOSED; |
| _orderlyClose.set(true); |
| break; |
| case CLOSE_RECEIVED: |
| case CLOSED: |
| break; |
| default: |
| throw new ServerScopedRuntimeException("Unknown state: " + _connectionState); |
| } |
| } |
| |
| private void closeReceived() |
| { |
| Collection<Session_1_0> sessions = new ArrayList<>(_sessions); |
| |
| for (final Session_1_0 session : sessions) |
| { |
| AccessController.doPrivileged((PrivilegedAction<Object>) () -> |
| { |
| session.remoteEnd(new End()); |
| return null; |
| }, session.getAccessControllerContext()); |
| } |
| } |
| |
| |
| @Override |
| public AMQPDescribedTypeRegistry getDescribedTypeRegistry() |
| { |
| return _describedTypeRegistry; |
| } |
| |
| @Override |
| public SectionDecoderRegistry getSectionDecoderRegistry() |
| { |
| return _describedTypeRegistry.getSectionDecoderRegistry(); |
| } |
| |
| @Override |
| public boolean isClosed() |
| { |
| return _connectionState == ConnectionState.CLOSED |
| || _connectionState == ConnectionState.CLOSE_RECEIVED; |
| } |
| |
| @Override |
| public boolean isClosing() |
| { |
| return _connectionState == ConnectionState.CLOSED |
| || _connectionState == ConnectionState.CLOSE_RECEIVED |
| || _connectionState == ConnectionState.CLOSE_SENT; |
| } |
| |
| @Override |
| public boolean closedForInput() |
| { |
| return _connectionState == ConnectionState.CLOSE_RECEIVED || _connectionState == ConnectionState.CLOSED ; |
| } |
| |
| @Override |
| public void sessionEnded(final Session_1_0 session) |
| { |
| _sessions.remove(session); |
| } |
| |
| private void inputClosed() |
| { |
| if (!closedForInput()) |
| { |
| FRAME_LOGGER.debug("RECV[{}] : {}", _remoteAddress, "Underlying connection closed"); |
| _connectionState = ConnectionState.CLOSED; |
| closeSender(); |
| closeReceived(); |
| } |
| } |
| |
| private void closeSender() |
| { |
| setClosedForOutput(true); |
| } |
| |
| @Override |
| public String getRemoteContainerId() |
| { |
| return _remoteContainerId; |
| } |
| |
| public boolean isOpen() |
| { |
| return _connectionState == ConnectionState.OPENED; |
| } |
| |
| @Override |
| public void sendEnd(final int channel, final End end, final boolean remove) |
| { |
| sendFrame(channel, end); |
| if (remove) |
| { |
| _sendingSessions[channel] = null; |
| } |
| } |
| |
| @Override |
| public void receiveEnd(final int channel, final End end) |
| { |
| assertState(ConnectionState.OPENED); |
| final Session_1_0 session = getSession(channel); |
| if (session != null) |
| { |
| _receivingSessions[channel] = null; |
| session.receiveEnd(end); |
| } |
| else |
| { |
| closeConnectionWithInvalidChannel(channel, end); |
| } |
| } |
| |
| private void closeConnectionWithInvalidChannel(final int channel, final FrameBody frame) |
| { |
| closeConnection(AmqpError.INVALID_FIELD, String.format("%s frame received on channel %d which is not mapped", frame.getClass().getSimpleName().toLowerCase(), channel)); |
| } |
| |
| @Override |
| public void receiveDisposition(final int channel, |
| final Disposition disposition) |
| { |
| assertState(ConnectionState.OPENED); |
| final Session_1_0 session = getSession(channel); |
| if (session != null) |
| { |
| session.receiveDisposition(disposition); |
| } |
| else |
| { |
| closeConnectionWithInvalidChannel(channel, disposition); |
| } |
| |
| } |
| |
| @Override |
| public void receiveBegin(final int receivingChannelId, final Begin begin) |
| { |
| |
| assertState(ConnectionState.OPENED); |
| if (begin.getRemoteChannel() != null) |
| { |
| closeConnection(ConnectionError.FRAMING_ERROR, |
| "BEGIN received on channel " |
| + receivingChannelId |
| + " with given remote-channel " |
| + begin.getRemoteChannel() |
| + ". Since the broker does not spontaneously start channels, this must be an error."); |
| |
| } |
| else // Peer requesting session creation |
| { |
| |
| if (_receivingSessions[receivingChannelId] == null) |
| { |
| int sendingChannelId = getFirstFreeChannel(); |
| if (sendingChannelId == -1) |
| { |
| closeConnection(ConnectionError.FRAMING_ERROR, |
| "BEGIN received on channel " |
| + receivingChannelId |
| + ". There are no free channels for the broker to respond on."); |
| } |
| else |
| { |
| Session_1_0 session = new Session_1_0(this, |
| begin, |
| sendingChannelId, |
| receivingChannelId, |
| getContextValue(Long.class, AMQPConnection_1_0.CONNECTION_SESSION_CREDIT_WINDOW_SIZE)); |
| session.create(); |
| |
| _receivingSessions[receivingChannelId] = session; |
| _sendingSessions[sendingChannelId] = session; |
| |
| Begin beginToSend = new Begin(); |
| beginToSend.setRemoteChannel(UnsignedShort.valueOf(receivingChannelId)); |
| beginToSend.setNextOutgoingId(session.getNextOutgoingId()); |
| beginToSend.setOutgoingWindow(session.getOutgoingWindow()); |
| beginToSend.setIncomingWindow(session.getIncomingWindow()); |
| sendFrame(sendingChannelId, beginToSend); |
| |
| synchronized (_blockingLock) |
| { |
| _sessions.add(session); |
| if (_blocking) |
| { |
| session.block(); |
| } |
| } |
| } |
| } |
| else |
| { |
| closeConnection(ConnectionError.FRAMING_ERROR, |
| "BEGIN received on channel " + receivingChannelId + " which is already in use."); |
| } |
| |
| } |
| |
| } |
| |
| private int getFirstFreeChannel() |
| { |
| for (int i = 0; i <= _channelMax; i++) |
| { |
| if (_sendingSessions[i] == null) |
| { |
| return i; |
| } |
| } |
| return -1; |
| } |
| |
| @Override |
| public void handleError(final Error error) |
| { |
| if (!_closedForOutput) |
| { |
| closeConnection(error); |
| } |
| } |
| |
| @Override |
| public void receiveTransfer(final int channel, final Transfer transfer) |
| { |
| assertState(ConnectionState.OPENED); |
| final Session_1_0 session = getSession(channel); |
| if (session != null) |
| { |
| session.receiveTransfer(transfer); |
| } |
| else |
| { |
| closeConnectionWithInvalidChannel(channel, transfer); |
| } |
| } |
| |
| @Override |
| public void receiveFlow(final int channel, final Flow flow) |
| { |
| assertState(ConnectionState.OPENED); |
| final Session_1_0 session = getSession(channel); |
| if (session != null) |
| { |
| session.receiveFlow(flow); |
| } |
| else |
| { |
| closeConnectionWithInvalidChannel(channel, flow); |
| } |
| |
| } |
| |
| @Override |
| public void receiveOpen(final int channel, final Open open) |
| { |
| assertState(ConnectionState.AWAIT_OPEN); |
| |
| int channelMax = getPort().getSessionCountLimit() - 1; |
| _channelMax = open.getChannelMax() == null ? channelMax |
| : open.getChannelMax().intValue() < channelMax |
| ? open.getChannelMax().intValue() |
| : channelMax; |
| if (_receivingSessions == null) |
| { |
| _receivingSessions = new Session_1_0[_channelMax + 1]; |
| _sendingSessions = new Session_1_0[_channelMax + 1]; |
| } |
| _maxFrameSize = open.getMaxFrameSize() == null |
| || open.getMaxFrameSize().longValue() > getBroker().getNetworkBufferSize() |
| ? getBroker().getNetworkBufferSize() |
| : open.getMaxFrameSize().intValue(); |
| _remoteContainerId = open.getContainerId(); |
| |
| if(open.getHostname() != null && !"".equals(open.getHostname().trim())) |
| { |
| _localHostname = open.getHostname(); |
| } |
| |
| if(_localHostname == null || "".equals(_localHostname.trim()) && getNetwork().getSelectedHost() != null) |
| { |
| _localHostname = getNetwork().getSelectedHost(); |
| } |
| if (open.getIdleTimeOut() != null) |
| { |
| _outgoingIdleTimeout = open.getIdleTimeOut().longValue(); |
| } |
| final Map<Symbol, Object> remoteProperties = open.getProperties() == null |
| ? Collections.emptyMap() |
| : Collections.unmodifiableMap(new LinkedHashMap<>(open.getProperties())); |
| _remoteDesiredCapabilities = open.getDesiredCapabilities() == null ? Collections.emptySet() : Sets.newHashSet(open.getDesiredCapabilities()); |
| if (remoteProperties.containsKey(Symbol.valueOf("product"))) |
| { |
| setClientProduct(remoteProperties.get(Symbol.valueOf("product")).toString()); |
| } |
| if (remoteProperties.containsKey(Symbol.valueOf("version"))) |
| { |
| setClientVersion(remoteProperties.get(Symbol.valueOf("version")).toString()); |
| } |
| setClientId(_remoteContainerId); |
| if (_remoteDesiredCapabilities.contains(SoleConnectionConnectionProperties.SOLE_CONNECTION_FOR_CONTAINER)) |
| { |
| if (remoteProperties != null && remoteProperties.containsKey(SOLE_CONNECTION_ENFORCEMENT_POLICY)) |
| { |
| try |
| { |
| _soleConnectionEnforcementPolicy = SoleConnectionEnforcementPolicy.valueOf(remoteProperties.get( |
| SOLE_CONNECTION_ENFORCEMENT_POLICY)); |
| } |
| catch (IllegalArgumentException e) |
| { |
| closeConnection(AmqpError.INVALID_FIELD, e.getMessage()); |
| return; |
| } |
| } |
| else |
| { |
| _soleConnectionEnforcementPolicy = SoleConnectionEnforcementPolicy.REFUSE_CONNECTION; |
| } |
| } |
| |
| if (_outgoingIdleTimeout != 0L && _outgoingIdleTimeout < MINIMUM_SUPPORTED_IDLE_TIMEOUT) |
| { |
| closeConnection(ConnectionError.CONNECTION_FORCED, |
| "Requested idle timeout of " |
| + _outgoingIdleTimeout |
| + " is too low. The minimum supported timeout is" |
| + MINIMUM_SUPPORTED_IDLE_TIMEOUT); |
| } |
| else |
| { |
| initialiseHeartbeating(_outgoingIdleTimeout / 2L, _incomingIdleTimeout); |
| final NamedAddressSpace addressSpace = getPort().getAddressSpace(_localHostname); |
| if (addressSpace == null) |
| { |
| closeConnection(AmqpError.NOT_FOUND, "Unknown hostname in connection open: '" + _localHostname + "'"); |
| } |
| else |
| { |
| receiveOpenInternal(addressSpace); |
| } |
| } |
| } |
| |
| private void receiveOpenInternal(final NamedAddressSpace addressSpace) |
| { |
| if (!addressSpace.isActive()) |
| { |
| final Error err = new Error(); |
| populateConnectionRedirect(addressSpace, err); |
| closeConnection(err); |
| } |
| else |
| { |
| if (AuthenticatedPrincipal.getOptionalAuthenticatedPrincipalFromSubject(getSubject()) == null) |
| { |
| closeConnection(AmqpError.NOT_ALLOWED, "Connection has not been authenticated"); |
| } |
| else |
| { |
| try |
| { |
| boolean registerSucceeded = addressSpace.registerConnection(this, (existingConnections, newConnection) -> |
| { |
| boolean proceedWithRegistration = true; |
| if (newConnection instanceof AMQPConnection_1_0Impl && !newConnection.isClosing()) |
| { |
| List<ListenableFuture<Void>> rescheduleFutures = new ArrayList<>(); |
| for (AMQPConnection<?> existingConnection : StreamSupport.stream(existingConnections.spliterator(), false) |
| .filter(con -> con instanceof AMQPConnection_1_0) |
| .filter(con -> !con.isClosing()) |
| .filter(con -> con.getRemoteContainerName().equals(newConnection.getRemoteContainerName())) |
| .collect(Collectors.toList())) |
| { |
| SoleConnectionEnforcementPolicy soleConnectionEnforcementPolicy = null; |
| if (((AMQPConnection_1_0Impl) existingConnection)._soleConnectionEnforcementPolicy |
| != null) |
| { |
| soleConnectionEnforcementPolicy = |
| ((AMQPConnection_1_0Impl) existingConnection)._soleConnectionEnforcementPolicy; |
| } |
| else if (((AMQPConnection_1_0Impl) newConnection)._soleConnectionEnforcementPolicy != null) |
| { |
| soleConnectionEnforcementPolicy = |
| ((AMQPConnection_1_0Impl) newConnection)._soleConnectionEnforcementPolicy; |
| } |
| if (SoleConnectionEnforcementPolicy.REFUSE_CONNECTION.equals(soleConnectionEnforcementPolicy)) |
| { |
| _properties.put(Symbol.valueOf("amqp:connection-establishment-failed"), true); |
| Error error = new Error(AmqpError.INVALID_FIELD, |
| String.format( |
| "Connection closed due to sole-connection-enforcement-policy '%s'", |
| soleConnectionEnforcementPolicy.toString())); |
| error.setInfo(Collections.singletonMap(Symbol.valueOf("invalid-field"), Symbol.valueOf("container-id"))); |
| newConnection.doOnIOThreadAsync(() -> ((AMQPConnection_1_0Impl) newConnection).closeConnection(error)); |
| proceedWithRegistration = false; |
| break; |
| } |
| else if (SoleConnectionEnforcementPolicy.CLOSE_EXISTING.equals(soleConnectionEnforcementPolicy)) |
| { |
| final Error error = new Error(AmqpError.RESOURCE_LOCKED, |
| String.format( |
| "Connection closed due to sole-connection-enforcement-policy '%s'", |
| soleConnectionEnforcementPolicy.toString())); |
| error.setInfo(Collections.singletonMap(Symbol.valueOf("sole-connection-enforcement"), true)); |
| rescheduleFutures.add(existingConnection.doOnIOThreadAsync( |
| () -> ((AMQPConnection_1_0Impl) existingConnection).closeConnection(error))); |
| proceedWithRegistration = false; |
| } |
| } |
| if (!rescheduleFutures.isEmpty()) |
| { |
| doAfter(allAsList(rescheduleFutures), () -> newConnection.doOnIOThreadAsync(() -> receiveOpenInternal(addressSpace))); |
| } |
| } |
| return proceedWithRegistration; |
| }); |
| |
| if (registerSucceeded) |
| { |
| setAddressSpace(addressSpace); |
| |
| if (!addressSpace.authoriseCreateConnection(this)) |
| { |
| closeConnection(AmqpError.NOT_ALLOWED, "Connection refused"); |
| } |
| else |
| { |
| switch (_connectionState) |
| { |
| case AWAIT_OPEN: |
| sendOpen(_channelMax, _maxFrameSize); |
| _connectionState = ConnectionState.OPENED; |
| break; |
| case CLOSE_SENT: |
| case CLOSED: |
| // already sent our close - probably due to an error |
| break; |
| default: |
| throw new ConnectionScopedRuntimeException(String.format( |
| "Unexpected state %s during connection open.", _connectionState)); |
| } |
| } |
| } |
| } |
| catch (VirtualHostUnavailableException | AccessControlException e) |
| { |
| closeConnection(AmqpError.NOT_ALLOWED, e.getMessage()); |
| } |
| } |
| } |
| } |
| |
| private void populateConnectionRedirect(final NamedAddressSpace addressSpace, final Error err) |
| { |
| final String redirectHost = addressSpace.getRedirectHost(getPort()); |
| |
| if(redirectHost == null) |
| { |
| err.setCondition(ConnectionError.CONNECTION_FORCED); |
| err.setDescription("Virtual host '" + _localHostname + "' is not active"); |
| } |
| else |
| { |
| err.setCondition(ConnectionError.REDIRECT); |
| String networkHost; |
| int port; |
| if(redirectHost.matches("\\[[0-9a-f:]+\\](:[0-9]+)?")) |
| { |
| // IPv6 case |
| networkHost = redirectHost.substring(1, redirectHost.indexOf("]")); |
| if(redirectHost.contains("]:")) |
| { |
| port = Integer.parseInt(redirectHost.substring(redirectHost.indexOf("]")+2)); |
| } |
| else |
| { |
| port = -1; |
| } |
| } |
| else |
| { |
| if(redirectHost.contains(":")) |
| { |
| networkHost = redirectHost.substring(0, redirectHost.lastIndexOf(":")); |
| try |
| { |
| String portString = redirectHost.substring(redirectHost.lastIndexOf(":")+1); |
| port = Integer.parseInt(portString); |
| } |
| catch (NumberFormatException e) |
| { |
| port = -1; |
| } |
| } |
| else |
| { |
| networkHost = redirectHost; |
| port = -1; |
| } |
| } |
| final Map<Symbol, Object> infoMap = new HashMap<>(); |
| infoMap.put(Symbol.valueOf("network-host"), networkHost); |
| if(port > 0) |
| { |
| infoMap.put(Symbol.valueOf("port"), UnsignedInteger.valueOf(port)); |
| } |
| err.setInfo(infoMap); |
| } |
| } |
| |
| @Override |
| public void receiveDetach(final int channel, final Detach detach) |
| { |
| assertState(ConnectionState.OPENED); |
| final Session_1_0 session = getSession(channel); |
| if (session != null) |
| { |
| session.receiveDetach(detach); |
| } |
| else |
| { |
| closeConnectionWithInvalidChannel(channel, detach); |
| } |
| } |
| |
| private void transportStateChanged() |
| { |
| for (Session_1_0 session : _sessions) |
| { |
| session.transportStateChanged(); |
| } |
| } |
| |
| @Override |
| public void close(final Error error) |
| { |
| closeConnection(error); |
| } |
| |
| private void setRemoteAddress(final SocketAddress remoteAddress) |
| { |
| _remoteAddress = remoteAddress; |
| } |
| |
| public void setOfferedCapabilities(final List<Symbol> offeredCapabilities) |
| { |
| _offeredCapabilities = offeredCapabilities; |
| } |
| |
| |
| private void setClosedForOutput(final boolean closed) |
| { |
| _closedForOutput = closed; |
| } |
| |
| @Override |
| public String getLocalFQDN() |
| { |
| return _localHostname != null ? _localHostname : super.getLocalFQDN(); |
| } |
| |
| @Override |
| public int getMaxFrameSize() |
| { |
| return _maxFrameSize; |
| } |
| |
| @Override |
| public int getChannelMax() |
| { |
| return _channelMax; |
| } |
| |
| @Override |
| public Object getReference() |
| { |
| return _reference; |
| } |
| |
| private void endpointClosed() |
| { |
| try |
| { |
| performDeleteTasks(); |
| closeReceived(); |
| } |
| finally |
| { |
| NamedAddressSpace virtualHost = getAddressSpace(); |
| if (virtualHost != null) |
| { |
| virtualHost.deregisterConnection(this); |
| } |
| } |
| } |
| |
| private void closeConnection(ErrorCondition errorCondition, String description) |
| { |
| closeConnection(new Error(errorCondition, description)); |
| } |
| |
| private void closeConnection(final Error error) |
| { |
| LOGGER.debug("Closing connection {} (state={}) due to {}", this, _connectionState, error); |
| _closeCause = error.getDescription(); |
| Close close = new Close(); |
| close.setError(error); |
| switch (_connectionState) |
| { |
| case AWAIT_AMQP_OR_SASL_HEADER: |
| case AWAIT_SASL_INIT: |
| case AWAIT_SASL_RESPONSE: |
| case AWAIT_AMQP_HEADER: |
| throw new ConnectionScopedRuntimeException("Connection is closed before being fully established: " + error.getDescription()); |
| |
| case AWAIT_OPEN: |
| sendOpen(0, 0); |
| sendClose(close); |
| _connectionState = ConnectionState.CLOSED; |
| getSender().close(); |
| break; |
| case OPENED: |
| sendClose(close); |
| _connectionState = ConnectionState.CLOSE_SENT; |
| addCloseTicker(); |
| break; |
| case CLOSE_RECEIVED: |
| sendClose(close); |
| _connectionState = ConnectionState.CLOSED; |
| addCloseTicker(); |
| break; |
| case CLOSE_SENT: |
| case CLOSED: |
| // already sent our close - too late to do anything more |
| break; |
| default: |
| throw new ServerScopedRuntimeException("Unknown state: " + _connectionState); |
| } |
| } |
| |
| @Override |
| public int sendFrame(final int channel, final FrameBody body, final QpidByteBuffer payload) |
| { |
| if (!_closedForOutput) |
| { |
| ValueWriter<FrameBody> writer = _describedTypeRegistry.getValueWriter(body); |
| if (payload == null) |
| { |
| send(new TransportFrame(channel, body)); |
| return 0; |
| } |
| else |
| { |
| int size = writer.getEncodedSize(); |
| int maxPayloadSize = _maxFrameSize - (size + 9); |
| long payloadLength = (long) payload.remaining(); |
| if (payloadLength <= maxPayloadSize) |
| { |
| send(new TransportFrame(channel, body, payload)); |
| return (int)payloadLength; |
| } |
| else |
| { |
| ((Transfer) body).setMore(Boolean.TRUE); |
| |
| writer = _describedTypeRegistry.getValueWriter(body); |
| size = writer.getEncodedSize(); |
| maxPayloadSize = _maxFrameSize - (size + 9); |
| |
| try (QpidByteBuffer payloadDup = payload.view(0, maxPayloadSize)) |
| { |
| payload.position(payload.position() + maxPayloadSize); |
| send(new TransportFrame(channel, body, payloadDup)); |
| } |
| |
| return maxPayloadSize; |
| } |
| } |
| } |
| else |
| { |
| return -1; |
| } |
| } |
| |
| @Override |
| public void sendFrame(final int channel, final FrameBody body) |
| { |
| sendFrame(channel, body, null); |
| } |
| |
| public ByteBufferSender getSender() |
| { |
| return getNetwork().getSender(); |
| } |
| |
| @Override |
| public void writerIdle() |
| { |
| send(TransportFrame.HEARTBEAT); |
| } |
| |
| @Override |
| public void readerIdle() |
| { |
| AccessController.doPrivileged((PrivilegedAction<Object>) () -> |
| { |
| getEventLogger().message(ConnectionMessages.IDLE_CLOSE("", false)); |
| getNetwork().close(); |
| return null; |
| }, getAccessControllerContext()); |
| } |
| |
| @Override |
| public void encryptedTransport() |
| { |
| } |
| |
| public String getAddress() |
| { |
| return getNetwork().getRemoteAddress().toString(); |
| } |
| |
| @Override |
| protected void onReceive(final QpidByteBuffer msg) |
| { |
| try |
| { |
| int remaining; |
| |
| try |
| { |
| do |
| { |
| remaining = msg.remaining(); |
| |
| switch (_connectionState) |
| { |
| case AWAIT_AMQP_OR_SASL_HEADER: |
| case AWAIT_AMQP_HEADER: |
| if (remaining >= 8) |
| { |
| processProtocolHeader(msg); |
| } |
| break; |
| case AWAIT_SASL_INIT: |
| case AWAIT_SASL_RESPONSE: |
| case AWAIT_OPEN: |
| case OPENED: |
| case CLOSE_SENT: |
| _frameHandler.parse(msg); |
| break; |
| case CLOSE_RECEIVED: |
| case CLOSED: |
| // ignore; |
| break; |
| } |
| |
| |
| } |
| while (msg.remaining() != remaining); |
| } |
| finally |
| { |
| receivedComplete(); |
| } |
| } |
| catch (IllegalArgumentException | IllegalStateException e) |
| { |
| throw new ConnectionScopedRuntimeException(e); |
| } |
| } |
| |
| @Override |
| public void receivedComplete() |
| { |
| if (_receivingSessions != null) |
| { |
| for (final Session_1_0 session : _receivingSessions) |
| { |
| if (session != null) |
| { |
| final AccessControlContext context = session.getAccessControllerContext(); |
| AccessController.doPrivileged((PrivilegedAction<Void>) () -> |
| { |
| session.receivedComplete(); |
| return null; |
| }, context); |
| } |
| } |
| } |
| } |
| |
| private void processProtocolHeader(final QpidByteBuffer msg) |
| { |
| if(msg.remaining() >= 8) |
| { |
| byte[] header = new byte[8]; |
| msg.get(header); |
| |
| final AuthenticationProvider<?> authenticationProvider = getPort().getAuthenticationProvider(); |
| |
| if(Arrays.equals(header, SASL_HEADER)) |
| { |
| if(_saslComplete) |
| { |
| throw new ConnectionScopedRuntimeException("SASL Layer header received after SASL already established"); |
| } |
| |
| try (QpidByteBuffer protocolHeader = QpidByteBuffer.wrap(SASL_HEADER)) |
| { |
| getSender().send(protocolHeader); |
| } |
| SaslMechanisms mechanisms = new SaslMechanisms(); |
| ArrayList<Symbol> mechanismsList = new ArrayList<>(); |
| for (String name : authenticationProvider.getAvailableMechanisms(getTransport().isSecure())) |
| { |
| mechanismsList.add(Symbol.valueOf(name)); |
| } |
| mechanisms.setSaslServerMechanisms(mechanismsList.toArray(new Symbol[mechanismsList.size()])); |
| send(new SASLFrame(mechanisms)); |
| |
| _connectionState = ConnectionState.AWAIT_SASL_INIT; |
| _frameHandler = getFrameHandler(true); |
| } |
| else if(Arrays.equals(header, AMQP_HEADER)) |
| { |
| if(!_saslComplete) |
| { |
| final List<String> mechanisms = authenticationProvider.getAvailableMechanisms(getTransport().isSecure()); |
| |
| if(mechanisms.contains(ExternalAuthenticationManagerImpl.MECHANISM_NAME) && getNetwork().getPeerPrincipal() != null) |
| { |
| setUserPrincipal(new AuthenticatedPrincipal(getNetwork().getPeerPrincipal())); |
| } |
| else if(mechanisms.contains(AnonymousAuthenticationManager.MECHANISM_NAME)) |
| { |
| setUserPrincipal(new AuthenticatedPrincipal(((AnonymousAuthenticationManager) authenticationProvider).getAnonymousPrincipal())); |
| } |
| else |
| { |
| LOGGER.warn("{} : attempt to initiate AMQP connection without correctly authenticating", getLogSubject()); |
| _connectionState = ConnectionState.CLOSED; |
| getNetwork().close(); |
| } |
| |
| } |
| try (QpidByteBuffer protocolHeader = QpidByteBuffer.wrap(AMQP_HEADER)) |
| { |
| getSender().send(protocolHeader); |
| } |
| _connectionState = ConnectionState.AWAIT_OPEN; |
| _frameHandler = getFrameHandler(false); |
| |
| } |
| else |
| { |
| LOGGER.warn("{} : unknown AMQP header {}", getLogSubject(), Functions.str(header)); |
| _connectionState = ConnectionState.CLOSED; |
| getNetwork().close(); |
| } |
| |
| } |
| |
| } |
| |
| private FrameHandler getFrameHandler(final boolean sasl) |
| { |
| return new FrameHandler(new ValueHandler(this.getDescribedTypeRegistry()), this, sasl); |
| } |
| |
| |
| @Override |
| public void closed() |
| { |
| try |
| { |
| inputClosed(); |
| } |
| catch(RuntimeException e) |
| { |
| LOGGER.error("Exception while closing", e); |
| } |
| finally |
| { |
| try |
| { |
| endpointClosed(); |
| } |
| finally |
| { |
| markTransportClosed(); |
| } |
| } |
| } |
| |
| private void send(final AMQFrame amqFrame) |
| { |
| updateLastWriteTime(); |
| FRAME_LOGGER.debug("SEND[{}|{}] : {}", |
| getNetwork().getRemoteAddress(), |
| amqFrame.getChannel(), |
| amqFrame.getFrameBody() == null ? "<<HEARTBEAT>>" : amqFrame.getFrameBody()); |
| |
| int size = _frameWriter.send(amqFrame); |
| if (size > getMaxFrameSize()) |
| { |
| throw new OversizeFrameException(amqFrame, size); |
| } |
| } |
| |
| private void addCloseTicker() |
| { |
| long timeoutTime = System.currentTimeMillis() + getContextValue(Long.class, Connection.CLOSE_RESPONSE_TIMEOUT); |
| |
| getAggregateTicker().addTicker(new ConnectionClosingTicker(timeoutTime, getNetwork())); |
| |
| // trigger a wakeup to ensure the ticker will be taken into account |
| notifyWork(); |
| } |
| |
| @Override |
| public boolean isTransportBlockedForWriting() |
| { |
| return _transportBlockedForWriting; |
| } |
| |
| @Override |
| public void setTransportBlockedForWriting(final boolean blocked) |
| { |
| if(_transportBlockedForWriting != blocked) |
| { |
| _transportBlockedForWriting = blocked; |
| transportStateChanged(); |
| } |
| |
| } |
| |
| @Override |
| public Iterator<Runnable> processPendingIterator() |
| { |
| if (isIOThread()) |
| { |
| return new ProcessPendingIterator(); |
| } |
| else |
| { |
| return Collections.emptyIterator(); |
| } |
| } |
| |
| @Override |
| public boolean hasWork() |
| { |
| return _stateChanged.get(); |
| } |
| |
| @Override |
| public void notifyWork() |
| { |
| _stateChanged.set(true); |
| |
| final Action<ProtocolEngine> listener = _workListener.get(); |
| if(listener != null) |
| { |
| listener.performAction(this); |
| } |
| } |
| |
| @Override |
| public void notifyWork(final AMQPSession<?,?> sessionModel) |
| { |
| _sessionsWithWork.add(sessionModel); |
| notifyWork(); |
| } |
| |
| @Override |
| public void clearWork() |
| { |
| _stateChanged.set(false); |
| } |
| |
| @Override |
| public void setWorkListener(final Action<ProtocolEngine> listener) |
| { |
| _workListener.set(listener); |
| } |
| |
| @Override |
| public boolean hasSessionWithName(final byte[] name) |
| { |
| return false; |
| } |
| |
| @Override |
| public void sendConnectionCloseAsync(final CloseReason reason, final String description) |
| { |
| |
| stopConnection(); |
| final ErrorCondition cause; |
| switch(reason) |
| { |
| case MANAGEMENT: |
| cause = ConnectionError.CONNECTION_FORCED; |
| break; |
| case TRANSACTION_TIMEOUT: |
| cause = AmqpError.RESOURCE_LIMIT_EXCEEDED; |
| break; |
| default: |
| cause = AmqpError.INTERNAL_ERROR; |
| } |
| Action<ConnectionHandler> action = object -> closeConnection(cause, description); |
| addAsyncTask(action); |
| } |
| |
| @Override |
| public void closeSessionAsync(final AMQPSession<?,?> session, |
| final CloseReason reason, final String message) |
| { |
| final ErrorCondition cause; |
| switch(reason) |
| { |
| case MANAGEMENT: |
| cause = ConnectionError.CONNECTION_FORCED; |
| break; |
| case TRANSACTION_TIMEOUT: |
| cause = AmqpError.RESOURCE_LIMIT_EXCEEDED; |
| break; |
| default: |
| cause = AmqpError.INTERNAL_ERROR; |
| } |
| addAsyncTask(object -> AccessController.doPrivileged(new PrivilegedAction<Void>() { |
| @Override |
| public Void run() |
| { |
| ((Session_1_0)session).close(cause, message); |
| return null; |
| } |
| }, ((Session_1_0)session).getAccessControllerContext())); |
| |
| } |
| |
| @Override |
| public void block() |
| { |
| synchronized (_blockingLock) |
| { |
| if (!_blocking) |
| { |
| _blocking = true; |
| doOnIOThreadAsync(this::doBlock); |
| } |
| } |
| } |
| |
| private void doBlock() |
| { |
| for(Session_1_0 session : _sessions) |
| { |
| session.block(); |
| } |
| } |
| |
| @Override |
| public String getRemoteContainerName() |
| { |
| return _remoteContainerId; |
| } |
| |
| @Override |
| public Collection<? extends Session_1_0> getSessionModels() |
| { |
| return Collections.unmodifiableCollection(_sessions); |
| } |
| |
| @Override |
| public void unblock() |
| { |
| synchronized (_blockingLock) |
| { |
| if(_blocking) |
| { |
| _blocking = false; |
| doOnIOThreadAsync(this::doUnblock); |
| } |
| } |
| } |
| |
| private void doUnblock() |
| { |
| for(Session_1_0 session : _sessions) |
| { |
| session.unblock(); |
| } |
| } |
| |
| @Override |
| public int getSessionCountLimit() |
| { |
| return _channelMax + 1; |
| } |
| |
| @Override |
| public boolean isOrderlyClose() |
| { |
| return _orderlyClose.get(); |
| } |
| |
| @Override |
| protected String getCloseCause() |
| { |
| return _closeCause; |
| } |
| |
| @Override |
| public boolean getSendSaslFinalChallengeAsChallenge() |
| { |
| return _sendSaslFinalChallengeAsChallenge; |
| } |
| |
| @Override |
| protected void addAsyncTask(final Action<? super ConnectionHandler> action) |
| { |
| _asyncTaskList.add(action); |
| notifyWork(); |
| } |
| |
| |
| private void sendOpen(final int channelMax, final int maxFrameSize) |
| { |
| Open open = new Open(); |
| |
| Map<String,Object> props = Collections.emptyMap(); |
| for(ConnectionPropertyEnricher enricher : getPort().getConnectionPropertyEnrichers()) |
| { |
| props = enricher.addConnectionProperties(this, props); |
| } |
| for(Map.Entry<String,Object> entry : props.entrySet()) |
| { |
| _properties.put(Symbol.valueOf(entry.getKey()), entry.getValue()); |
| } |
| |
| if (_receivingSessions == null) |
| { |
| _receivingSessions = new Session_1_0[channelMax + 1]; |
| _sendingSessions = new Session_1_0[channelMax + 1]; |
| } |
| if (channelMax < _channelMax) |
| { |
| _channelMax = channelMax; |
| } |
| open.setChannelMax(UnsignedShort.valueOf((short) channelMax)); |
| open.setContainerId(getAddressSpace() == null ? UUID.randomUUID().toString() : getAddressSpace().getId().toString()); |
| open.setMaxFrameSize(UnsignedInteger.valueOf(maxFrameSize)); |
| // TODO - should we try to set the hostname based on the connection information? |
| // open.setHostname(); |
| open.setIdleTimeOut(UnsignedInteger.valueOf(_incomingIdleTimeout)); |
| |
| // set the offered capabilities |
| if(_offeredCapabilities != null && !_offeredCapabilities.isEmpty()) |
| { |
| open.setOfferedCapabilities(_offeredCapabilities.toArray(new Symbol[_offeredCapabilities.size()])); |
| } |
| |
| if (_remoteDesiredCapabilities != null |
| && _remoteDesiredCapabilities.contains(SoleConnectionConnectionProperties.SOLE_CONNECTION_FOR_CONTAINER)) |
| { |
| _properties.put(SoleConnectionConnectionProperties.SOLE_CONNECTION_DETECTION_POLICY, |
| SoleConnectionDetectionPolicy.STRONG); |
| } |
| |
| if (_soleConnectionEnforcementPolicy == SoleConnectionEnforcementPolicy.CLOSE_EXISTING) |
| { |
| _properties.put(SOLE_CONNECTION_ENFORCEMENT_POLICY, SoleConnectionEnforcementPolicy.CLOSE_EXISTING.getValue()); |
| } |
| |
| open.setProperties(_properties); |
| |
| sendFrame(CONNECTION_CONTROL_CHANNEL, open); |
| } |
| |
| private Session_1_0 getSession(final int channel) |
| { |
| Session_1_0 session = _receivingSessions[channel]; |
| if (session == null) |
| { |
| Error error = new Error(); |
| error.setCondition(ConnectionError.FRAMING_ERROR); |
| error.setDescription("Frame received on channel " + channel + " which is not known as a begun session."); |
| handleError(error); |
| } |
| |
| return session; |
| } |
| |
| private void sendClose(Close closeToSend) |
| { |
| sendFrame(CONNECTION_CONTROL_CHANNEL, closeToSend); |
| closeSender(); |
| } |
| |
| |
| private void assertState(final ConnectionState state) |
| { |
| if (_connectionState != state) |
| { |
| throw new ConnectionScopedRuntimeException(String.format( |
| "Unexpected state, client has sent frame in an illegal order. Required state: %s, actual state: %s", |
| state, |
| _connectionState)); |
| } |
| } |
| |
| private class ProcessPendingIterator implements Iterator<Runnable> |
| { |
| private Iterator<? extends AMQPSession<?,?>> _sessionIterator; |
| private ProcessPendingIterator() |
| { |
| _sessionIterator = _sessionsWithWork.iterator(); |
| } |
| |
| @Override |
| public boolean hasNext() |
| { |
| return (!_sessionsWithWork.isEmpty() && !isClosed() && !isConnectionStopped()) || !_asyncTaskList.isEmpty(); |
| } |
| |
| @Override |
| public Runnable next() |
| { |
| if(!_sessionsWithWork.isEmpty()) |
| { |
| if(isClosed() || isConnectionStopped()) |
| { |
| |
| final Action<? super ConnectionHandler> asyncAction = _asyncTaskList.poll(); |
| if(asyncAction != null) |
| { |
| return () -> asyncAction.performAction(AMQPConnection_1_0Impl.this); |
| } |
| else |
| { |
| return () -> { }; |
| } |
| } |
| else |
| { |
| if (!_sessionIterator.hasNext()) |
| { |
| _sessionIterator = _sessionsWithWork.iterator(); |
| } |
| final AMQPSession<?,?> session = _sessionIterator.next(); |
| return () -> |
| { |
| _sessionIterator.remove(); |
| if (session.processPending()) |
| { |
| _sessionsWithWork.add(session); |
| } |
| }; |
| } |
| } |
| else if(!_asyncTaskList.isEmpty()) |
| { |
| final Action<? super ConnectionHandler> asyncAction = _asyncTaskList.poll(); |
| return () -> asyncAction.performAction(AMQPConnection_1_0Impl.this); |
| } |
| else |
| { |
| throw new NoSuchElementException(); |
| } |
| } |
| |
| @Override |
| public void remove() |
| { |
| throw new UnsupportedOperationException(); |
| } |
| } |
| |
| @Override |
| public Iterator<ServerTransaction> getOpenTransactions() |
| { |
| return _openTransactions.values().iterator(); |
| } |
| |
| @Override |
| public IdentifiedTransaction createIdentifiedTransaction() |
| { |
| final LocalTransaction serverTransaction = createLocalTransaction(); |
| int id = 0; |
| while (id >= 0 && _openTransactions.putIfAbsent(id, serverTransaction) != null) |
| { |
| id++; |
| } |
| if (id < 0) |
| { |
| throw new IllegalStateException("Unsupported state, too many opened transactions"); |
| } |
| return new IdentifiedTransaction(id, serverTransaction); |
| } |
| |
| @Override |
| public ServerTransaction getTransaction(final int txnId) |
| { |
| return Optional.ofNullable(_openTransactions.get(txnId)) |
| .orElseThrow(() -> new UnknownTransactionException(txnId)); |
| } |
| |
| @Override |
| public void removeTransaction(final int txnId) |
| { |
| Optional.ofNullable(_openTransactions.remove(txnId)) |
| .orElseThrow(() -> new UnknownTransactionException(txnId)); |
| } |
| |
| @Override |
| protected boolean isOpeningInProgress() |
| { |
| switch (_connectionState) |
| { |
| case AWAIT_AMQP_OR_SASL_HEADER: |
| case AWAIT_SASL_INIT: |
| case AWAIT_SASL_RESPONSE: |
| case AWAIT_AMQP_HEADER: |
| case AWAIT_OPEN: |
| return true; |
| case OPENED: |
| case CLOSE_RECEIVED: |
| case CLOSE_SENT: |
| case CLOSED: |
| return false; |
| default: |
| throw new IllegalStateException("Unsupported state " + _connectionState); |
| } |
| } |
| } |
