| /* |
| * |
| * 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.v0_10; |
| |
| import static org.apache.qpid.server.protocol.v0_10.ServerSession.State.CLOSED; |
| import static org.apache.qpid.server.protocol.v0_10.ServerSession.State.CLOSING; |
| import static org.apache.qpid.server.protocol.v0_10.ServerSession.State.DETACHED; |
| import static org.apache.qpid.server.protocol.v0_10.ServerSession.State.NEW; |
| import static org.apache.qpid.server.protocol.v0_10.ServerSession.State.OPEN; |
| import static org.apache.qpid.server.protocol.v0_10.ServerSession.State.RESUMING; |
| import static org.apache.qpid.server.protocol.v0_10.transport.Option.COMPLETED; |
| import static org.apache.qpid.server.protocol.v0_10.transport.Option.TIMELY_REPLY; |
| import static org.apache.qpid.server.util.Serial.ge; |
| import static org.apache.qpid.server.util.Serial.gt; |
| import static org.apache.qpid.server.util.Serial.le; |
| import static org.apache.qpid.server.util.Serial.lt; |
| import static org.apache.qpid.server.util.Serial.max; |
| import static org.apache.qpid.server.util.Strings.toUTF8; |
| |
| import java.nio.ByteBuffer; |
| import java.security.AccessControlContext; |
| import java.security.AccessController; |
| import java.security.Principal; |
| import java.security.PrivilegedAction; |
| import java.util.ArrayList; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.Deque; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.SortedMap; |
| import java.util.UUID; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.ConcurrentLinkedDeque; |
| import java.util.concurrent.ConcurrentSkipListMap; |
| import java.util.concurrent.Semaphore; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| import java.util.concurrent.atomic.AtomicInteger; |
| import java.util.concurrent.atomic.AtomicReference; |
| |
| import javax.security.auth.Subject; |
| |
| import com.google.common.util.concurrent.ListenableFuture; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import org.apache.qpid.server.model.Session; |
| import org.apache.qpid.server.txn.AsyncCommand; |
| import org.apache.qpid.server.logging.LogMessage; |
| import org.apache.qpid.server.logging.LogSubject; |
| import org.apache.qpid.server.logging.messages.ChannelMessages; |
| import org.apache.qpid.server.message.InstanceProperties; |
| import org.apache.qpid.server.message.MessageDestination; |
| import org.apache.qpid.server.message.MessageInstance; |
| import org.apache.qpid.server.message.MessageInstanceConsumer; |
| import org.apache.qpid.server.message.RoutingResult; |
| import org.apache.qpid.server.model.Broker; |
| import org.apache.qpid.server.model.NamedAddressSpace; |
| import org.apache.qpid.server.model.Queue; |
| import org.apache.qpid.server.protocol.v0_10.transport.*; |
| import org.apache.qpid.server.store.MessageStore; |
| import org.apache.qpid.server.store.StoreException; |
| import org.apache.qpid.server.transport.AMQPConnection; |
| import org.apache.qpid.server.txn.AlreadyKnownDtxException; |
| import org.apache.qpid.server.txn.AsyncAutoCommitTransaction; |
| import org.apache.qpid.server.txn.DistributedTransaction; |
| import org.apache.qpid.server.txn.DtxNotSelectedException; |
| import org.apache.qpid.server.txn.IncorrectDtxStateException; |
| import org.apache.qpid.server.txn.JoinAndResumeDtxException; |
| import org.apache.qpid.server.txn.LocalTransaction; |
| import org.apache.qpid.server.txn.NotAssociatedDtxException; |
| import org.apache.qpid.server.txn.RollbackOnlyDtxException; |
| import org.apache.qpid.server.txn.ServerTransaction; |
| import org.apache.qpid.server.txn.SuspendAndFailDtxException; |
| import org.apache.qpid.server.txn.TimeoutDtxException; |
| import org.apache.qpid.server.txn.UnknownDtxBranchException; |
| import org.apache.qpid.server.util.Action; |
| |
| public class ServerSession extends SessionInvoker |
| implements LogSubject, AsyncAutoCommitTransaction.FutureRecorder |
| { |
| private static final Logger LOGGER = LoggerFactory.getLogger(ServerSession.class); |
| public static final int UNLIMITED_CREDIT = 0xFFFFFFFF; |
| |
| private static final String NULL_DESTINATION = UUID.randomUUID().toString(); |
| private static final int PRODUCER_CREDIT_TOPUP_THRESHOLD = 1 << 30; |
| private static final int UNFINISHED_COMMAND_QUEUE_THRESHOLD = 500; |
| |
| private final Set<Object> _blockingEntities = Collections.synchronizedSet(new HashSet<>()); |
| private final Deque<AsyncCommand> _unfinishedCommandsQueue = new ConcurrentLinkedDeque<>(); |
| |
| private final AtomicBoolean _blocking = new AtomicBoolean(false); |
| private final AtomicInteger _outstandingCredit = new AtomicInteger(UNLIMITED_CREDIT); |
| // completed incoming commands |
| private final Object processedLock = new Object(); |
| private final int commandLimit = Integer.getInteger("qpid.session.command_limit", 64 * 1024); |
| private final Object commandsLock = new Object(); |
| private final Object stateLock = new Object(); |
| private final Map<String, ConsumerTarget_0_10> _subscriptions = new ConcurrentHashMap<>(); |
| private final AtomicReference<LogMessage> _forcedCloseLogMessage = new AtomicReference<>(); |
| private final long _blockingTimeout; |
| private final ServerConnection connection; |
| private final Binary name; |
| private final ServerSessionDelegate delegate; |
| private final Map<Integer,Method> commands = new HashMap<>(); |
| private final int byteLimit = Integer.getInteger("qpid.session.byte_limit", 1024 * 1024); |
| private final Semaphore credit = new Semaphore(0); |
| private final Map<Integer,ResultFuture<?>> results = new HashMap<>(); |
| private final SortedMap<Integer, MessageDispositionChangeListener> _messageDispositionListenerMap = |
| new ConcurrentSkipListMap<>(); |
| |
| private Session_0_10 _modelObject; |
| private long _blockTime; |
| private boolean _wireBlockingState; |
| private boolean closing; |
| private int channel; |
| private boolean incomingInit; |
| // incoming command count |
| private int commandsIn; |
| private RangeSet processed; |
| private int maxProcessed; |
| private int syncPoint; |
| // outgoing command count |
| private int commandsOut = 0; |
| private int commandBytes = 0; |
| private int maxComplete = commandsOut - 1; |
| private State state = NEW; |
| private Thread resumer = null; |
| private boolean transacted = false; |
| private SessionDetachCode detachCode; |
| private boolean _isNoReplay = false; |
| private org.apache.qpid.server.protocol.v0_10.transport.ExecutionException exception = null; |
| private volatile ServerTransaction _transaction; |
| |
| public ServerSession(ServerConnection connection, ServerSessionDelegate delegate, Binary name, long expiry) |
| { |
| this.connection = connection; |
| this.delegate = delegate; |
| this.name = name; |
| this.closing = false; |
| this._isNoReplay = false; |
| initReceiver(); |
| _transaction = new AsyncAutoCommitTransaction(this.getMessageStore(),this); |
| |
| _blockingTimeout = connection.getBroker().getContextValue(Long.class, Broker.CHANNEL_FLOW_CONTROL_ENFORCEMENT_TIMEOUT); |
| } |
| |
| public Binary getName() |
| { |
| return name; |
| } |
| |
| protected void setClose(boolean close) |
| { |
| this.closing = close; |
| } |
| |
| public int getChannel() |
| { |
| return channel; |
| } |
| |
| void setChannel(int channel) |
| { |
| this.channel = channel; |
| } |
| |
| protected State getState() |
| { |
| return this.state; |
| } |
| |
| void addCredit(int value) |
| { |
| credit.release(value); |
| } |
| |
| void drainCredit() |
| { |
| credit.drainPermits(); |
| } |
| |
| private void initReceiver() |
| { |
| synchronized (processedLock) |
| { |
| incomingInit = false; |
| processed = RangeSetFactory.createRangeSet(); |
| } |
| } |
| |
| void attach() |
| { |
| initReceiver(); |
| sessionAttach(name.getBytes()); |
| sessionRequestTimeout(0);//use expiry here only if/when session resume is supported |
| } |
| |
| void resume() |
| { |
| synchronized (commandsLock) |
| { |
| attach(); |
| |
| for (int i = maxComplete + 1; lt(i, commandsOut); i++) |
| { |
| Method m = getCommand(i); |
| if (m == null) |
| { |
| m = new ExecutionSync(); |
| m.setId(i); |
| } |
| else if (m instanceof MessageTransfer) |
| { |
| MessageTransfer xfr = (MessageTransfer)m; |
| |
| Header header = xfr.getHeader(); |
| |
| if (header != null) |
| { |
| if (header.getDeliveryProperties() != null) |
| { |
| header.getDeliveryProperties().setRedelivered(true); |
| } |
| else |
| { |
| DeliveryProperties deliveryProps = new DeliveryProperties(); |
| deliveryProps.setRedelivered(true); |
| |
| xfr.setHeader(new Header(deliveryProps, header.getMessageProperties(), |
| header.getNonStandardProperties())); |
| } |
| |
| } |
| else |
| { |
| DeliveryProperties deliveryProps = new DeliveryProperties(); |
| deliveryProps.setRedelivered(true); |
| xfr.setHeader(new Header(deliveryProps, null, null)); |
| } |
| } |
| sessionCommandPoint(m.getId(), 0); |
| send(m); |
| } |
| |
| sessionCommandPoint(commandsOut, 0); |
| |
| sessionFlush(COMPLETED); |
| resumer = Thread.currentThread(); |
| state = RESUMING; |
| |
| if(isTransacted()) |
| { |
| txSelect(); |
| } |
| |
| resumer = null; |
| } |
| } |
| |
| private Method getCommand(int i) |
| { |
| return commands.get(i); |
| } |
| |
| private void setCommand(int commandId, Method command) |
| { |
| commands.put(commandId, command); |
| } |
| |
| private Method removeCommand(int id) |
| { |
| return commands.remove(id); |
| } |
| |
| final void commandPoint(int id) |
| { |
| synchronized (processedLock) |
| { |
| this.commandsIn = id; |
| if (!incomingInit) |
| { |
| incomingInit = true; |
| maxProcessed = commandsIn - 1; |
| syncPoint = maxProcessed; |
| } |
| } |
| } |
| |
| public int getCommandsOut() |
| { |
| return commandsOut; |
| } |
| |
| public int getCommandsIn() |
| { |
| return commandsIn; |
| } |
| |
| public int nextCommandId() |
| { |
| return commandsIn++; |
| } |
| |
| final void identify(Method cmd) |
| { |
| if (!incomingInit) |
| { |
| throw new IllegalStateException(); |
| } |
| |
| int id = nextCommandId(); |
| cmd.setId(id); |
| |
| if(LOGGER.isDebugEnabled()) |
| { |
| LOGGER.debug("identify: ch={}, commandId={}", this.channel, id); |
| } |
| |
| if ((id & 0xff) == 0) |
| { |
| flushProcessed(TIMELY_REPLY); |
| } |
| } |
| |
| public void processed(Method command) |
| { |
| processed(command.getId()); |
| } |
| |
| public void processed(int command) |
| { |
| processed(command, command); |
| } |
| |
| public void processed(Range range) |
| { |
| |
| processed(range.getLower(), range.getUpper()); |
| } |
| |
| public void processed(int lower, int upper) |
| { |
| if(LOGGER.isDebugEnabled()) |
| { |
| LOGGER.debug("{} ch={} processed([{},{}]) {} {}", this, channel, lower, upper, syncPoint, maxProcessed); |
| } |
| |
| boolean flush; |
| synchronized (processedLock) |
| { |
| if(LOGGER.isDebugEnabled()) |
| { |
| LOGGER.debug("{} processed: {}", this, processed); |
| } |
| |
| if (ge(upper, commandsIn)) |
| { |
| throw new IllegalArgumentException |
| ("range exceeds max received command-id: " + Range.newInstance(lower, upper)); |
| } |
| |
| processed.add(lower, upper); |
| |
| Range first = processed.getFirst(); |
| |
| int flower = first.getLower(); |
| int fupper = first.getUpper(); |
| int old = maxProcessed; |
| if (le(flower, maxProcessed + 1)) |
| { |
| maxProcessed = max(maxProcessed, fupper); |
| } |
| boolean synced = ge(maxProcessed, syncPoint); |
| flush = lt(old, syncPoint) && synced; |
| if (synced) |
| { |
| syncPoint = maxProcessed; |
| } |
| } |
| if (flush) |
| { |
| flushProcessed(); |
| } |
| } |
| |
| void flushExpected() |
| { |
| RangeSet rs = RangeSetFactory.createRangeSet(); |
| synchronized (processedLock) |
| { |
| if (incomingInit) |
| { |
| rs.add(commandsIn); |
| } |
| } |
| sessionExpected(rs, null); |
| } |
| |
| public void flushProcessed(Option... options) |
| { |
| RangeSet copy; |
| synchronized (processedLock) |
| { |
| copy = processed.copy(); |
| } |
| |
| synchronized (commandsLock) |
| { |
| if (state == DETACHED || state == CLOSING || state == CLOSED) |
| { |
| return; |
| } |
| if (copy.size() > 0) |
| { |
| sessionCompleted(copy, options); |
| } |
| } |
| } |
| |
| void knownComplete(RangeSet kc) |
| { |
| if (kc.size() > 0) |
| { |
| synchronized (processedLock) |
| { |
| processed.subtract(kc) ; |
| } |
| } |
| } |
| |
| void syncPoint() |
| { |
| int id = getCommandsIn() - 1; |
| LOGGER.debug("{} synced to {}", this, id); |
| boolean flush; |
| synchronized (processedLock) |
| { |
| syncPoint = id; |
| flush = ge(maxProcessed, syncPoint); |
| } |
| if (flush) |
| { |
| flushProcessed(); |
| } |
| } |
| |
| protected boolean complete(int lower, int upper) |
| { |
| //avoid autoboxing |
| if(LOGGER.isDebugEnabled()) |
| { |
| LOGGER.debug("{} complete({}, {})", this, lower, upper); |
| } |
| synchronized (commandsLock) |
| { |
| int old = maxComplete; |
| for (int id = max(maxComplete, lower); le(id, upper); id++) |
| { |
| Method m = removeCommand(id); |
| if (m != null) |
| { |
| commandBytes -= m.getBodySize(); |
| m.complete(); |
| } |
| } |
| if (le(lower, maxComplete + 1)) |
| { |
| maxComplete = max(maxComplete, upper); |
| } |
| |
| if(LOGGER.isDebugEnabled()) |
| { |
| LOGGER.debug("{} commands remaining: {}", this, commandsOut - maxComplete); |
| } |
| |
| commandsLock.notifyAll(); |
| return gt(maxComplete, old); |
| } |
| } |
| |
| void received(Method m) |
| { |
| m.delegate(this, delegate); |
| } |
| |
| private void send(Method m) |
| { |
| m.setChannel(channel); |
| connection.send(m); |
| |
| if (!m.isBatch()) |
| { |
| connection.flush(); |
| } |
| } |
| |
| protected boolean isBytesFull() |
| { |
| return commandBytes >= byteLimit; |
| } |
| |
| protected boolean isCommandsFull(int id) |
| { |
| return id - maxComplete >= commandLimit; |
| } |
| |
| @Override |
| public void invoke(Method m) |
| { |
| invoke(m,(Runnable)null); |
| } |
| |
| public void invoke(Method m, Runnable postIdSettingAction) |
| { |
| if (m.getEncodedTrack() == Frame.L4) |
| { |
| synchronized (commandsLock) |
| { |
| if (state == DETACHED && m.isUnreliable()) |
| { |
| Thread current = Thread.currentThread(); |
| if (!current.equals(resumer)) |
| { |
| return; |
| } |
| } |
| |
| if (state != OPEN && state != CLOSED && state != CLOSING) |
| { |
| Thread current = Thread.currentThread(); |
| if (!current.equals(resumer) ) |
| { |
| // Should not happen |
| throw new SessionException(String.format("Unexpected state %s", state)); |
| } |
| } |
| |
| switch (state) |
| { |
| case OPEN: |
| break; |
| case RESUMING: |
| Thread current = Thread.currentThread(); |
| if (!current.equals(resumer)) |
| { |
| throw new SessionException |
| ("timed out waiting for resume to finish"); |
| } |
| break; |
| case CLOSING: |
| case CLOSED: |
| org.apache.qpid.server.protocol.v0_10.transport.ExecutionException exc = getException(); |
| if (exc != null) |
| { |
| throw new SessionException(exc); |
| } |
| else |
| { |
| throw new SessionClosedException(); |
| } |
| default: |
| throw new SessionException |
| (String.format |
| ("timed out waiting for session to become open " + |
| "(state=%s)", state)); |
| } |
| |
| int next; |
| next = commandsOut++; |
| m.setId(next); |
| if(postIdSettingAction != null) |
| { |
| postIdSettingAction.run(); |
| } |
| |
| if (isFull(next)) |
| { |
| // Should not happen |
| throw new SessionException(String.format("Command buffer full next: %d", next)); |
| } |
| |
| if (state == CLOSED) |
| { |
| org.apache.qpid.server.protocol.v0_10.transport.ExecutionException exc = getException(); |
| if (exc != null) |
| { |
| throw new SessionException(exc); |
| } |
| else |
| { |
| throw new SessionClosedException(); |
| } |
| } |
| |
| if (isFull(next)) |
| { |
| throw new SessionException("timed out waiting for completion"); |
| } |
| |
| if (next == 0) |
| { |
| sessionCommandPoint(0, 0); |
| } |
| |
| boolean replayTransfer = !_isNoReplay && !closing && !transacted && |
| m instanceof MessageTransfer && |
| ! m.isUnreliable(); |
| |
| if ((replayTransfer) || m.hasCompletionListener()) |
| { |
| setCommand(next, m); |
| commandBytes += m.getBodySize(); |
| } |
| |
| try |
| { |
| send(m); |
| } |
| catch (SenderException e) |
| { |
| if (!closing) |
| { |
| // if we are not closing then this will happen |
| // again on resume |
| LOGGER.error("error sending command", e); |
| } |
| else |
| { |
| e.rethrow(); |
| } |
| } |
| |
| // flush every 64K commands to avoid ambiguity on |
| // wraparound |
| if (shouldIssueFlush(next)) |
| { |
| try |
| { |
| sessionFlush(COMPLETED); |
| } |
| catch (SenderException e) |
| { |
| if (!closing) |
| { |
| // if expiry is > 0 then this will happen |
| // again on resume |
| LOGGER.error("error sending flush (periodic)", e); |
| } |
| else |
| { |
| e.rethrow(); |
| } |
| } |
| } |
| } |
| } |
| else |
| { |
| send(m); |
| } |
| } |
| |
| protected boolean shouldIssueFlush(int next) |
| { |
| return (next % 65536) == 0; |
| } |
| |
| void result(int command, Struct result) |
| { |
| ResultFuture<?> future; |
| synchronized (results) |
| { |
| future = results.remove(command); |
| } |
| |
| if (future != null) |
| { |
| future.set(result); |
| } |
| else |
| { |
| LOGGER.warn("Received a response to a command" + |
| " that's no longer valid on the client side." + |
| " [ command id : {} , result : {} ]", command, result); |
| } |
| } |
| |
| void setException(org.apache.qpid.server.protocol.v0_10.transport.ExecutionException exc) |
| { |
| synchronized (results) |
| { |
| if (exception != null) |
| { |
| throw new IllegalStateException( |
| String.format("too many exceptions: %s, %s", exception, exc)); |
| } |
| exception = exc; |
| } |
| } |
| |
| org.apache.qpid.server.protocol.v0_10.transport.ExecutionException getException() |
| { |
| synchronized (results) |
| { |
| return exception; |
| } |
| } |
| |
| @Override |
| protected <T> Future<T> invoke(Method m, Class<T> klass) |
| { |
| synchronized (commandsLock) |
| { |
| int command = commandsOut; |
| ResultFuture<T> future = new ResultFuture<>(klass); |
| synchronized (results) |
| { |
| results.put(command, future); |
| } |
| invoke(m); |
| return future; |
| } |
| } |
| |
| public final void messageTransfer(String destination, |
| MessageAcceptMode acceptMode, |
| MessageAcquireMode acquireMode, |
| Header header, |
| byte[] body, |
| Option ... _options) { |
| messageTransfer(destination, acceptMode, acquireMode, header, |
| ByteBuffer.wrap(body), _options); |
| } |
| |
| public final void messageTransfer(String destination, |
| MessageAcceptMode acceptMode, |
| MessageAcquireMode acquireMode, |
| Header header, |
| String body, |
| Option ... _options) { |
| messageTransfer(destination, acceptMode, acquireMode, header, |
| toUTF8(body), _options); |
| } |
| |
| public void exception(Throwable t) |
| { |
| LOGGER.error("caught exception", t); |
| } |
| |
| public void closed() |
| { |
| synchronized (commandsLock) |
| { |
| if (closing || getException() != null) |
| { |
| state = CLOSED; |
| } |
| else |
| { |
| state = DETACHED; |
| } |
| |
| commandsLock.notifyAll(); |
| |
| synchronized (results) |
| { |
| for (ResultFuture<?> result : results.values()) |
| { |
| synchronized(result) |
| { |
| result.notifyAll(); |
| } |
| } |
| } |
| if(state == CLOSED) |
| { |
| delegate.closed(this); |
| } |
| else |
| { |
| delegate.detached(this); |
| } |
| } |
| |
| if(state == CLOSED) |
| { |
| connection.removeSession(this); |
| } |
| } |
| |
| public boolean isClosing() |
| { |
| return state == CLOSED || state == CLOSING || connection.isClosing(); |
| } |
| |
| @Override |
| public String toString() |
| { |
| return String.format("ssn:%s", name); |
| } |
| |
| public void setTransacted(boolean b) { |
| this.transacted = b; |
| } |
| |
| public boolean isTransacted(){ |
| return transacted; |
| } |
| |
| public void setDetachCode(SessionDetachCode dtc) |
| { |
| this.detachCode = dtc; |
| } |
| |
| public SessionDetachCode getDetachCode() |
| { |
| return this.detachCode; |
| } |
| |
| public Object getStateLock() |
| { |
| return stateLock; |
| } |
| |
| protected void sendSessionAttached(final byte[] name, final Option... options) |
| { |
| super.sessionAttached(name, options); |
| } |
| |
| public enum State { NEW, DETACHED, RESUMING, OPEN, CLOSING, CLOSED } |
| |
| public interface MessageDispositionChangeListener |
| { |
| void onAccept(); |
| |
| void onRelease(boolean setRedelivered, final boolean closing); |
| |
| void onReject(); |
| |
| boolean acquire(); |
| } |
| |
| private interface MessageDispositionAction |
| { |
| void performAction(MessageDispositionChangeListener listener); |
| } |
| |
| public Subject getSubject() |
| { |
| return _modelObject.getSubject(); |
| } |
| |
| public AccessControlContext getAccessControllerContext() |
| { |
| return _modelObject.getAccessControllerContext(); |
| } |
| |
| protected void setState(final State state) |
| { |
| if(runningAsSubject()) |
| { |
| synchronized (commandsLock) |
| { |
| this.state = state; |
| commandsLock.notifyAll(); |
| } |
| |
| if (state == State.OPEN) |
| { |
| getAMQPConnection().getEventLogger().message(ChannelMessages.CREATE()); |
| } |
| } |
| else |
| { |
| runAsSubject(() -> |
| { |
| setState(state); |
| return null; |
| }); |
| |
| } |
| } |
| |
| |
| private <T> T runAsSubject(final PrivilegedAction<T> privilegedAction) |
| { |
| return AccessController.doPrivileged(privilegedAction, getAccessControllerContext()); |
| } |
| |
| private boolean runningAsSubject() |
| { |
| return getAuthorizedSubject().equals(Subject.getSubject(AccessController.getContext())); |
| } |
| |
| private void invokeBlock() |
| { |
| invoke(new MessageSetFlowMode("", MessageFlowMode.CREDIT)); |
| invoke(new MessageStop("")); |
| } |
| |
| private void invokeUnblock() |
| { |
| MessageFlow mf = new MessageFlow(); |
| mf.setUnit(MessageCreditUnit.MESSAGE); |
| mf.setDestination(""); |
| _outstandingCredit.set(Integer.MAX_VALUE); |
| mf.setValue(Integer.MAX_VALUE); |
| invoke(mf); |
| } |
| |
| void authorisePublish(final MessageDestination destination, |
| final String routingKey, |
| final boolean immediate, |
| final long currentTime) |
| { |
| _modelObject.getPublishAuthCache().authorisePublish(destination, routingKey, immediate, currentTime); |
| } |
| |
| protected boolean isFull(int id) |
| { |
| return isCommandsFull(id); |
| } |
| |
| RoutingResult<MessageTransferMessage> enqueue(final MessageTransferMessage message, |
| final InstanceProperties instanceProperties, |
| final MessageDestination exchange) |
| { |
| if(_outstandingCredit.get() != UNLIMITED_CREDIT |
| && _outstandingCredit.decrementAndGet() == (Integer.MAX_VALUE - PRODUCER_CREDIT_TOPUP_THRESHOLD)) |
| { |
| _outstandingCredit.addAndGet(PRODUCER_CREDIT_TOPUP_THRESHOLD); |
| invoke(new MessageFlow("",MessageCreditUnit.MESSAGE, PRODUCER_CREDIT_TOPUP_THRESHOLD)); |
| } |
| // locally cache arrival time to ensure that we don't reload metadata |
| final long arrivalTime = message.getArrivalTime(); |
| final RoutingResult<MessageTransferMessage> result = |
| exchange.route(message, message.getInitialRoutingAddress(), instanceProperties); |
| result.send(_transaction, null); |
| |
| getModelObject().registerMessageReceived(message.getSize()); |
| if (isTransactional()) |
| { |
| getModelObject().registerTransactedMessageReceived(); |
| } |
| return result; |
| } |
| |
| public void sendMessage(MessageTransfer xfr, |
| Runnable postIdSettingAction) |
| { |
| getModelObject().registerMessageDelivered(xfr.getBodySize()); |
| if (_transaction.isTransactional()) |
| { |
| getModelObject().registerTransactedMessageDelivered(); |
| } |
| invoke(xfr, postIdSettingAction); |
| } |
| |
| |
| public void onMessageDispositionChange(MessageTransfer xfr, MessageDispositionChangeListener acceptListener) |
| { |
| _messageDispositionListenerMap.put(xfr.getId(), acceptListener); |
| } |
| |
| public void accept(RangeSet ranges) |
| { |
| dispositionChange(ranges, MessageDispositionChangeListener::onAccept); |
| } |
| |
| |
| public void release(RangeSet ranges, final boolean setRedelivered) |
| { |
| dispositionChange(ranges, listener -> listener.onRelease(setRedelivered, false)); |
| } |
| |
| public void reject(RangeSet ranges) |
| { |
| dispositionChange(ranges, MessageDispositionChangeListener::onReject); |
| } |
| |
| public RangeSet acquire(RangeSet transfers) |
| { |
| RangeSet acquired = RangeSetFactory.createRangeSet(); |
| |
| if(!_messageDispositionListenerMap.isEmpty()) |
| { |
| Iterator<Integer> unacceptedMessages = _messageDispositionListenerMap.keySet().iterator(); |
| Iterator<Range> rangeIter = transfers.iterator(); |
| |
| if(rangeIter.hasNext()) |
| { |
| Range range = rangeIter.next(); |
| |
| while(range != null && unacceptedMessages.hasNext()) |
| { |
| int next = unacceptedMessages.next(); |
| while(gt(next, range.getUpper())) |
| { |
| if(rangeIter.hasNext()) |
| { |
| range = rangeIter.next(); |
| } |
| else |
| { |
| range = null; |
| break; |
| } |
| } |
| if(range != null && range.includes(next)) |
| { |
| MessageDispositionChangeListener changeListener = _messageDispositionListenerMap.get(next); |
| if(changeListener != null && changeListener.acquire()) |
| { |
| acquired.add(next); |
| } |
| } |
| |
| |
| } |
| |
| } |
| |
| |
| } |
| |
| return acquired; |
| } |
| |
| public void dispositionChange(RangeSet ranges, MessageDispositionAction action) |
| { |
| if(ranges != null) |
| { |
| |
| if(ranges.size() == 1) |
| { |
| Range r = ranges.getFirst(); |
| for(int i = r.getLower(); i <= r.getUpper(); i++) |
| { |
| MessageDispositionChangeListener changeListener = _messageDispositionListenerMap.remove(i); |
| if(changeListener != null) |
| { |
| action.performAction(changeListener); |
| } |
| } |
| } |
| else if(!_messageDispositionListenerMap.isEmpty()) |
| { |
| Iterator<Integer> unacceptedMessages = _messageDispositionListenerMap.keySet().iterator(); |
| Iterator<Range> rangeIter = ranges.iterator(); |
| |
| if(rangeIter.hasNext()) |
| { |
| Range range = rangeIter.next(); |
| |
| while(range != null && unacceptedMessages.hasNext()) |
| { |
| int next = unacceptedMessages.next(); |
| while(gt(next, range.getUpper())) |
| { |
| if(rangeIter.hasNext()) |
| { |
| range = rangeIter.next(); |
| } |
| else |
| { |
| range = null; |
| break; |
| } |
| } |
| if(range != null && range.includes(next)) |
| { |
| MessageDispositionChangeListener changeListener = _messageDispositionListenerMap.remove(next); |
| action.performAction(changeListener); |
| } |
| |
| |
| } |
| |
| } |
| } |
| } |
| } |
| |
| public void removeDispositionListener(Method method) |
| { |
| _messageDispositionListenerMap.remove(method.getId()); |
| } |
| |
| public void onClose() |
| { |
| AMQPConnection_0_10 amqpConnection = getAMQPConnection(); |
| if(_transaction instanceof LocalTransaction) |
| { |
| if (((LocalTransaction) _transaction).hasOutstandingWork()) |
| { |
| amqpConnection.incrementTransactionRollbackCounter(); |
| } |
| amqpConnection.decrementTransactionOpenCounter(); |
| _transaction.rollback(); |
| |
| amqpConnection.unregisterTransactionTickers(_transaction); |
| } |
| else if(_transaction instanceof DistributedTransaction) |
| { |
| getAddressSpace().getDtxRegistry().endAssociations(_modelObject); |
| } |
| |
| for(MessageDispositionChangeListener listener : _messageDispositionListenerMap.values()) |
| { |
| listener.onRelease(true, true); |
| } |
| _messageDispositionListenerMap.clear(); |
| |
| for (Action<? super Session_0_10> task : _modelObject.getTaskList()) |
| { |
| task.performAction(_modelObject); |
| } |
| |
| LogMessage operationalLoggingMessage = _forcedCloseLogMessage.get(); |
| if (operationalLoggingMessage == null && getConnection().getConnectionCloseMessage() != null) |
| { |
| operationalLoggingMessage = ChannelMessages.CLOSE_FORCED(getConnection().getConnectionCloseCode(), |
| getConnection().getConnectionCloseMessage()); |
| } |
| if (operationalLoggingMessage == null) |
| { |
| operationalLoggingMessage = ChannelMessages.CLOSE(); |
| } |
| amqpConnection.getEventLogger().message(getLogSubject(), operationalLoggingMessage); |
| } |
| |
| protected void awaitClose() |
| { |
| // Broker shouldn't block awaiting close - thus do override this method to do nothing |
| } |
| |
| public void acknowledge(final MessageInstanceConsumer consumer, |
| final ConsumerTarget_0_10 target, |
| final MessageInstance entry) |
| { |
| if (entry.makeAcquisitionUnstealable(consumer)) |
| { |
| _transaction.dequeue(entry.getEnqueueRecord(), |
| new ServerTransaction.Action() |
| { |
| |
| @Override |
| public void postCommit() |
| { |
| entry.delete(); |
| } |
| |
| @Override |
| public void onRollback() |
| { |
| // The client has acknowledge the message and therefore have seen it. |
| // In the event of rollback, the message must be marked as redelivered. |
| entry.setRedelivered(); |
| entry.release(consumer); |
| } |
| }); |
| } |
| } |
| |
| Collection<ConsumerTarget_0_10> getSubscriptions() |
| { |
| return _subscriptions.values(); |
| } |
| |
| public void register(String destination, ConsumerTarget_0_10 sub) |
| { |
| _subscriptions.put(destination == null ? NULL_DESTINATION : destination, sub); |
| } |
| |
| |
| public ConsumerTarget_0_10 getSubscription(String destination) |
| { |
| return _subscriptions.get(destination == null ? NULL_DESTINATION : destination); |
| } |
| |
| public void unregister(ConsumerTarget_0_10 sub) |
| { |
| _subscriptions.remove(sub.getName()); |
| sub.close(); |
| |
| } |
| |
| public boolean isTransactional() |
| { |
| return _transaction.isTransactional(); |
| } |
| |
| ServerTransaction getTransaction() |
| { |
| return _transaction; |
| } |
| |
| public void selectTx() |
| { |
| ServerTransaction txn = _transaction; |
| AMQPConnection_0_10 amqpConnection = getAMQPConnection(); |
| if (txn instanceof LocalTransaction) |
| { |
| amqpConnection.unregisterTransactionTickers(_transaction); |
| } |
| |
| _transaction = amqpConnection.createLocalTransaction(); |
| long notificationRepeatPeriod = |
| getModelObject().getContextValue(Long.class, Session.TRANSACTION_TIMEOUT_NOTIFICATION_REPEAT_PERIOD); |
| amqpConnection.registerTransactionTickers(_transaction, |
| message -> amqpConnection.sendConnectionCloseAsync(AMQPConnection.CloseReason.TRANSACTION_TIMEOUT, |
| (String) message), |
| notificationRepeatPeriod); |
| } |
| |
| public void selectDtx() |
| { |
| _transaction = new DistributedTransaction(_modelObject, getAddressSpace().getDtxRegistry()); |
| |
| } |
| |
| |
| public void startDtx(Xid xid, boolean join, boolean resume) |
| throws JoinAndResumeDtxException, |
| UnknownDtxBranchException, |
| AlreadyKnownDtxException, |
| DtxNotSelectedException |
| { |
| DistributedTransaction distributedTransaction = assertDtxTransaction(); |
| distributedTransaction.start(toDtxXid(xid), join, resume); |
| } |
| |
| |
| public void endDtx(Xid xid, boolean fail, boolean suspend) |
| throws NotAssociatedDtxException, |
| UnknownDtxBranchException, |
| DtxNotSelectedException, |
| SuspendAndFailDtxException, TimeoutDtxException |
| { |
| DistributedTransaction distributedTransaction = assertDtxTransaction(); |
| distributedTransaction.end(toDtxXid(xid), fail, suspend); |
| } |
| |
| |
| public long getTimeoutDtx(Xid xid) |
| throws UnknownDtxBranchException |
| { |
| return getAddressSpace().getDtxRegistry().getTimeout(toDtxXid(xid)); |
| } |
| |
| |
| public void setTimeoutDtx(Xid xid, long timeout) |
| throws UnknownDtxBranchException |
| { |
| getAddressSpace().getDtxRegistry().setTimeout(toDtxXid(xid), timeout); |
| } |
| |
| |
| public void prepareDtx(Xid xid) |
| throws UnknownDtxBranchException, |
| IncorrectDtxStateException, StoreException, RollbackOnlyDtxException, TimeoutDtxException |
| { |
| getAddressSpace().getDtxRegistry().prepare(toDtxXid(xid)); |
| } |
| |
| public void commitDtx(Xid xid, boolean onePhase) |
| throws UnknownDtxBranchException, |
| IncorrectDtxStateException, StoreException, RollbackOnlyDtxException, TimeoutDtxException |
| { |
| getAddressSpace().getDtxRegistry().commit(toDtxXid(xid), onePhase); |
| } |
| |
| |
| public void rollbackDtx(Xid xid) |
| throws UnknownDtxBranchException, |
| IncorrectDtxStateException, StoreException, TimeoutDtxException |
| { |
| getAddressSpace().getDtxRegistry().rollback(toDtxXid(xid)); |
| } |
| |
| |
| public void forgetDtx(Xid xid) throws UnknownDtxBranchException, IncorrectDtxStateException |
| { |
| getAddressSpace().getDtxRegistry().forget(toDtxXid(xid)); |
| } |
| |
| public List<Xid> recoverDtx() |
| { |
| List<Xid> xids = new ArrayList<>(); |
| Iterator<org.apache.qpid.server.txn.Xid> dtxXids = getAddressSpace().getDtxRegistry().recover().iterator(); |
| while(dtxXids.hasNext()) |
| { |
| org.apache.qpid.server.txn.Xid dtxXid = dtxXids.next(); |
| xids.add(new Xid(dtxXid.getFormat(), dtxXid.getGlobalId(), dtxXid.getBranchId())); |
| } |
| return xids; |
| } |
| |
| private DistributedTransaction assertDtxTransaction() throws DtxNotSelectedException |
| { |
| if(_transaction instanceof DistributedTransaction) |
| { |
| return (DistributedTransaction) _transaction; |
| } |
| else |
| { |
| throw new DtxNotSelectedException(); |
| } |
| } |
| |
| |
| public void commit() |
| { |
| _transaction.commit(); |
| getAMQPConnection().incrementTransactionBeginCounter(); |
| } |
| |
| public void rollback() |
| { |
| _transaction.rollback(); |
| AMQPConnection_0_10 amqpConnection = getAMQPConnection(); |
| amqpConnection.incrementTransactionRollbackCounter(); |
| amqpConnection.incrementTransactionBeginCounter(); |
| } |
| |
| public int getChannelId() |
| { |
| return getChannel(); |
| } |
| |
| public Principal getAuthorizedPrincipal() |
| { |
| return getConnection().getAuthorizedPrincipal(); |
| } |
| |
| public Subject getAuthorizedSubject() |
| { |
| return getSubject(); |
| } |
| |
| public Object getReference() |
| { |
| return getConnection().getReference(); |
| } |
| |
| public MessageStore getMessageStore() |
| { |
| return getAddressSpace().getMessageStore(); |
| } |
| |
| public NamedAddressSpace getAddressSpace() |
| { |
| return getConnection().getAddressSpace(); |
| } |
| |
| public boolean isDurable() |
| { |
| return false; |
| } |
| |
| |
| public UUID getId() |
| { |
| return _modelObject.getId(); |
| } |
| |
| public AMQPConnection_0_10 getAMQPConnection() |
| { |
| return getConnection().getAmqpConnection(); |
| } |
| |
| public ServerConnection getConnection() |
| { |
| return connection; |
| } |
| |
| public LogSubject getLogSubject() |
| { |
| return _modelObject.getLogSubject(); |
| } |
| |
| public void block(Queue<?> queue) |
| { |
| block(queue, queue.getName()); |
| } |
| |
| public void block() |
| { |
| block(this, "** All Queues **"); |
| } |
| |
| |
| private void block(final Object queue, final String name) |
| { |
| synchronized (_blockingEntities) |
| { |
| if(_blockingEntities.add(queue)) |
| { |
| |
| if(_blocking.compareAndSet(false,true)) |
| { |
| getAMQPConnection().getEventLogger().message(getLogSubject(), ChannelMessages.FLOW_ENFORCED(name)); |
| if(getState() == State.OPEN) |
| { |
| getAMQPConnection().notifyWork(_modelObject); |
| } |
| } |
| |
| |
| } |
| } |
| } |
| |
| public void unblock(Queue<?> queue) |
| { |
| unblock((Object)queue); |
| } |
| |
| public void unblock() |
| { |
| unblock(this); |
| } |
| |
| private void unblock(final Object queue) |
| { |
| if(_blockingEntities.remove(queue) && _blockingEntities.isEmpty()) |
| { |
| if(_blocking.compareAndSet(true,false) && !isClosing()) |
| { |
| getAMQPConnection().getEventLogger().message(getLogSubject(), ChannelMessages.FLOW_REMOVED()); |
| getAMQPConnection().notifyWork(_modelObject); |
| } |
| } |
| } |
| |
| |
| boolean blockingTimeoutExceeded() |
| { |
| long blockTime = _blockTime; |
| boolean b = _wireBlockingState && blockTime != 0 && (System.currentTimeMillis() - blockTime) > _blockingTimeout; |
| return b; |
| } |
| |
| public void updateBlockedStateIfNecesssary() |
| { |
| boolean desiredBlockingState = _blocking.get(); |
| if (desiredBlockingState != _wireBlockingState) |
| { |
| _wireBlockingState = desiredBlockingState; |
| |
| if (desiredBlockingState) |
| { |
| invokeBlock(); |
| } |
| else |
| { |
| invokeUnblock(); |
| } |
| _blockTime = desiredBlockingState ? System.currentTimeMillis() : 0; |
| } |
| |
| } |
| |
| public Object getConnectionReference() |
| { |
| return getConnection().getReference(); |
| } |
| |
| @Override |
| public String toLogString() |
| { |
| return _modelObject.toLogString(); |
| } |
| |
| public void close(int cause, String message) |
| { |
| _forcedCloseLogMessage.compareAndSet(null, ChannelMessages.CLOSE_FORCED(cause, message)); |
| close(); |
| } |
| |
| public void close() |
| { |
| // unregister subscriptions in order to prevent sending of new messages |
| // to subscriptions with closing session |
| unregisterSubscriptions(); |
| if(_modelObject != null) |
| { |
| _modelObject.delete(); |
| } |
| if (LOGGER.isDebugEnabled()) |
| { |
| LOGGER.debug("Closing [{}] in state [{}]", this, state); |
| } |
| synchronized (commandsLock) |
| { |
| switch(state) |
| { |
| case DETACHED: |
| state = CLOSED; |
| delegate.closed(this); |
| connection.removeSession(this); |
| break; |
| case CLOSED: |
| break; |
| default: |
| state = CLOSING; |
| setClose(true); |
| sessionRequestTimeout(0); |
| sessionDetach(name.getBytes()); |
| awaitClose(); |
| } |
| } |
| } |
| |
| void unregisterSubscriptions() |
| { |
| final Collection<ConsumerTarget_0_10> subscriptions = getSubscriptions(); |
| for (ConsumerTarget_0_10 subscription_0_10 : subscriptions) |
| { |
| unregister(subscription_0_10); |
| } |
| } |
| |
| void stopSubscriptions() |
| { |
| final Collection<ConsumerTarget_0_10> subscriptions = getSubscriptions(); |
| for (ConsumerTarget_0_10 subscription_0_10 : subscriptions) |
| { |
| subscription_0_10.stop(); |
| } |
| } |
| |
| |
| void receivedComplete() |
| { |
| runAsSubject(() -> |
| { |
| final Collection<ConsumerTarget_0_10> subscriptions = getSubscriptions(); |
| for (ConsumerTarget_0_10 subscription_0_10 : subscriptions) |
| { |
| subscription_0_10.flushCreditState(false); |
| } |
| awaitCommandCompletion(); |
| return null; |
| }); |
| } |
| |
| public int getUnacknowledgedMessageCount() |
| { |
| return _messageDispositionListenerMap.size(); |
| } |
| |
| public boolean getBlocking() |
| { |
| return _blocking.get(); |
| } |
| |
| public void completeAsyncCommands() |
| { |
| AsyncCommand cmd; |
| while((cmd = _unfinishedCommandsQueue.peek()) != null && cmd.isReadyForCompletion()) |
| { |
| cmd.complete(); |
| _unfinishedCommandsQueue.poll(); |
| } |
| while(_unfinishedCommandsQueue.size() > UNFINISHED_COMMAND_QUEUE_THRESHOLD) |
| { |
| cmd = _unfinishedCommandsQueue.poll(); |
| cmd.complete(); |
| } |
| } |
| |
| |
| public void awaitCommandCompletion() |
| { |
| AsyncCommand cmd; |
| while((cmd = _unfinishedCommandsQueue.poll()) != null) |
| { |
| cmd.complete(); |
| } |
| } |
| |
| |
| public Object getAsyncCommandMark() |
| { |
| return _unfinishedCommandsQueue.isEmpty() ? null : _unfinishedCommandsQueue.getLast(); |
| } |
| |
| @Override |
| public void recordFuture(final ListenableFuture<Void> future, final ServerTransaction.Action action) |
| { |
| _unfinishedCommandsQueue.add(new AsyncCommand(future, action)); |
| } |
| |
| public void setModelObject(final Session_0_10 session) |
| { |
| _modelObject = session; |
| } |
| |
| public Session_0_10 getModelObject() |
| { |
| return _modelObject; |
| } |
| |
| public long getTransactionStartTimeLong() |
| { |
| ServerTransaction serverTransaction = _transaction; |
| if (serverTransaction.isTransactional()) |
| { |
| return serverTransaction.getTransactionStartTime(); |
| } |
| else |
| { |
| return 0L; |
| } |
| } |
| |
| public long getTransactionUpdateTimeLong() |
| { |
| ServerTransaction serverTransaction = _transaction; |
| if (serverTransaction.isTransactional()) |
| { |
| return serverTransaction.getTransactionUpdateTime(); |
| } |
| else |
| { |
| return 0L; |
| } |
| } |
| |
| private class ResultFuture<T> implements Future<T> |
| { |
| |
| private final Class<T> klass; |
| private T result; |
| |
| private ResultFuture(Class<T> klass) |
| { |
| this.klass = klass; |
| } |
| |
| private void set(Struct result) |
| { |
| synchronized (this) |
| { |
| this.result = klass.cast(result); |
| notifyAll(); |
| } |
| } |
| |
| @Override |
| public T get(long timeout) |
| { |
| synchronized (this) |
| { |
| Waiter w = new Waiter(this, timeout); |
| while (w.hasTime() && state != CLOSED && !isDone()) |
| { |
| LOGGER.debug("{} waiting for result: {}", ServerSession.this, this); |
| w.await(); |
| } |
| } |
| |
| if (isDone()) |
| { |
| return result; |
| } |
| else if (state == CLOSED) |
| { |
| org.apache.qpid.server.protocol.v0_10.transport.ExecutionException ex = getException(); |
| if(ex == null) |
| { |
| throw new SessionClosedException(); |
| } |
| throw new SessionException(ex); |
| } |
| else |
| { |
| throw new SessionException( |
| String.format("%s timed out waiting for result: %s", |
| ServerSession.this, this)); |
| } |
| } |
| |
| @Override |
| public boolean isDone() |
| { |
| return result != null; |
| } |
| |
| @Override |
| public String toString() |
| { |
| return String.format("Future(%s)", isDone() ? result : klass); |
| } |
| |
| } |
| |
| public static org.apache.qpid.server.txn.Xid toDtxXid(final Xid xid) |
| { |
| return new org.apache.qpid.server.txn.Xid(xid.getFormat(), |
| xid.getGlobalId(), |
| xid.getBranchId()); |
| } |
| |
| |
| } |
