You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kw...@apache.org on 2017/02/21 06:36:12 UTC
svn commit: r1783832 - in /qpid/java/trunk:
broker-core/src/main/java/org/apache/qpid/server/transport/
broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/
Author: kwall
Date: Tue Feb 21 06:36:12 2017
New Revision: 1783832
URL: http://svn.apache.org/viewvc?rev=1783832&view=rev
Log:
QPID-7622: [Java Broker] [0-10] Remove dead waiter and client failover hooks from ServerSession
Added:
qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/Future.java
- copied, changed from r1783342, qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/Future.java
Removed:
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/Future.java
Modified:
qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java
Copied: qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/Future.java (from r1783342, qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/Future.java)
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/Future.java?p2=qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/Future.java&p1=qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/Future.java&r1=1783342&r2=1783832&rev=1783832&view=diff
==============================================================================
--- qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/Future.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/Future.java Tue Feb 21 06:36:12 2017
@@ -1,5 +1,4 @@
/*
- *
* 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
@@ -18,7 +17,7 @@
* under the License.
*
*/
-package org.apache.qpid.server.transport;
+package org.apache.qpid.server.protocol.v0_10;
/**
@@ -30,7 +29,7 @@ package org.apache.qpid.server.transport
public interface Future<T>
{
- T get();
+ T get(long timeout);
boolean isDone();
Modified: qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java?rev=1783832&r1=1783831&r2=1783832&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java Tue Feb 21 06:36:12 2017
@@ -28,7 +28,6 @@ import static org.apache.qpid.server.pro
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.transport.Option.COMPLETED;
-import static org.apache.qpid.server.transport.Option.SYNC;
import static org.apache.qpid.server.transport.Option.TIMELY_REPLY;
import static org.apache.qpid.server.util.Serial.ge;
import static org.apache.qpid.server.util.Serial.gt;
@@ -140,7 +139,6 @@ public class ServerSession extends Sessi
private int channel;
private ServerSessionDelegate delegate;
private SessionListener listener = new DefaultSessionListener();
- private boolean autoSync = false;
private boolean incomingInit;
// incoming command count
private int commandsIn;
@@ -153,7 +151,6 @@ public class ServerSession extends Sessi
private int commandBytes = 0;
private int byteLimit = Integer.getInteger("qpid.session.byte_limit", 1024 * 1024);
private int maxComplete = commandsOut - 1;
- private boolean needSync = false;
private State state = NEW;
private Semaphore credit = new Semaphore(0);
private Thread resumer = null;
@@ -557,12 +554,8 @@ public class ServerSession extends Sessi
Thread current = Thread.currentThread();
if (!current.equals(resumer) )
{
- Waiter w = new Waiter(commandsLock, timeout);
- while (w.hasTime() && (state != OPEN && state != CLOSED))
- {
- checkFailoverRequired("Command was interrupted because of failover, before being sent");
- w.await();
- }
+ // Should not happen
+ throw new SessionException(String.format("Unexpected state %s", state));
}
}
@@ -606,32 +599,8 @@ public class ServerSession extends Sessi
if (isFull(next))
{
- Waiter w = new Waiter(commandsLock, timeout);
- while (w.hasTime() && isFull(next) && state != CLOSED)
- {
- if (state == OPEN || state == RESUMING)
- {
- try
- {
- sessionFlush(COMPLETED);
- }
- catch (SenderException e)
- {
- if (!closing)
- {
- // if expiry is > 0 then this will
- // happen again on resume
- LOGGER.error("error sending flush (full replay buffer)", e);
- }
- else
- {
- e.rethrow();
- }
- }
- }
- checkFailoverRequired("Command was interrupted because of failover, before being sent");
- w.await();
- }
+ // Should not happen
+ throw new SessionException(String.format("Command buffer full next: %d", next));
}
if (state == CLOSED)
@@ -666,11 +635,6 @@ public class ServerSession extends Sessi
setCommand(next, m);
commandBytes += m.getBodySize();
}
- if (autoSync)
- {
- m.setSync(true);
- }
- needSync = !m.isSync();
try
{
@@ -689,10 +653,6 @@ public class ServerSession extends Sessi
e.rethrow();
}
}
- if (autoSync)
- {
- sync();
- }
// flush every 64K commands to avoid ambiguity on
// wraparound
@@ -737,54 +697,6 @@ public class ServerSession extends Sessi
return (next % 65536) == 0;
}
- public void sync()
- {
- sync(timeout);
- }
-
- public void sync(long timeout)
- {
- LOGGER.debug("{} sync()", this);
- synchronized (commandsLock)
- {
- int point = commandsOut - 1;
-
- if (needSync && lt(maxComplete, point))
- {
- executionSync(SYNC);
- }
-
- Waiter w = new Waiter(commandsLock, timeout);
- while (w.hasTime() && state != CLOSED && lt(maxComplete, point))
- {
- checkFailoverRequired("Session sync was interrupted by failover.");
- if(LOGGER.isDebugEnabled())
- {
- LOGGER.debug("{} waiting for[{}]: {}, {}", this, point, maxComplete, commands);
- }
- w.await();
- }
-
- if (lt(maxComplete, point))
- {
- if (state != CLOSED)
- {
- throw new SessionException(
- String.format("timed out waiting for sync: complete = %s, point = %s",
- maxComplete, point));
- }
- else
- {
- org.apache.qpid.server.transport.ExecutionException ee = getException();
- if (ee != null)
- {
- throw new SessionException(ee);
- }
- }
- }
- }
- }
-
void result(int command, Struct result)
{
ResultFuture<?> future;
@@ -2049,16 +1961,12 @@ public class ServerSession extends Sessi
}
}
- public T get()
- {
- return get(timeout);
- }
-
public boolean isDone()
{
return result != null;
}
+ @Override
public String toString()
{
return String.format("Future(%s)", isDone() ? result : klass);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org