Fix for issue #2765 relative to unstable trafficshaping test procedure
Motivation:
The test procedure is unstable due to not enough precise timestamping
during the check.
Modifications:
Reducing the test cases and cibling "stable" test ("timestamp-able")
bring more stability to the tests.
Result:
Tests for TrafficShapingHandler seem more stable (whatever using JVM 6,
7 or 8).
This commit is contained in:
Trustin Lee
parent
44d9163169
commit
98dfc91471
@ -49,20 +49,27 @@ import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
import static org.junit.Assert.*;
|
||||
|
||||
public class TrafficShapingTest extends AbstractSocketTest {
|
||||
private static final InternalLogger logger = InternalLoggerFactory.getInstance(TrafficShapingTest.class);
|
||||
public class TrafficShapingHandlerTest extends AbstractSocketTest {
|
||||
private static final InternalLogger logger = InternalLoggerFactory.getInstance(TrafficShapingHandlerTest.class);
|
||||
private static final InternalLogger loggerServer = InternalLoggerFactory.getInstance("ServerTSH");
|
||||
private static final InternalLogger loggerClient = InternalLoggerFactory.getInstance("ClientTSH");
|
||||
|
||||
static final int messageSize = 1024;
|
||||
static final int bandwidthFactor = 15;
|
||||
static final int minfactor = bandwidthFactor - (bandwidthFactor / 2) - 1;
|
||||
static final int bandwidthFactor = 12;
|
||||
static final int minfactor = 3;
|
||||
static final int maxfactor = bandwidthFactor + (bandwidthFactor / 2);
|
||||
static final long stepms = 1000 / bandwidthFactor;
|
||||
static final long minimalms = Math.max(stepms / 2, 20) / 10 * 10;
|
||||
static final long check = Math.max(Math.min(100, minimalms / 2) / 10 * 10, 20);
|
||||
static final long check = 10;
|
||||
private static final Random random = new Random();
|
||||
static final byte[] data = new byte[messageSize];
|
||||
|
||||
private static final String TRAFFIC = "traffic";
|
||||
private static String TESTNAME;
|
||||
private static int TESTRUN;
|
||||
|
||||
private static EventExecutorGroup group;
|
||||
private static EventExecutorGroup groupForGlobal;
|
||||
private static ScheduledExecutorService executor = Executors.newScheduledThreadPool(10);
|
||||
static {
|
||||
random.nextBytes(data);
|
||||
@ -70,17 +77,20 @@ public class TrafficShapingTest extends AbstractSocketTest {
|
||||
|
||||
@BeforeClass
|
||||
public static void createGroup() {
|
||||
logger.info("Bandwidth: " + minfactor + " <= " + bandwidthFactor + " <= " + maxfactor +
|
||||
" StepMs: " + stepms + " MinMs: " + minimalms + " CheckMs: " + check);
|
||||
InternalLoggerFactory.setDefaultFactory(new Slf4JLoggerFactory());
|
||||
Logger logger = (Logger) LoggerFactory.getLogger("ROOT");
|
||||
logger.setLevel(Level.INFO);
|
||||
logger.info("Bandwidth: " + minfactor + " <= " + bandwidthFactor + " <= " + maxfactor +
|
||||
" StepMs: " + stepms + " MinMs: " + minimalms + " CheckMs: " + check);
|
||||
group = new DefaultEventExecutorGroup(8);
|
||||
groupForGlobal = new DefaultEventExecutorGroup(8);
|
||||
}
|
||||
|
||||
@AfterClass
|
||||
public static void destroyGroup() throws Exception {
|
||||
group.shutdownGracefully().sync();
|
||||
groupForGlobal.shutdownGracefully().sync();
|
||||
executor.shutdown();
|
||||
}
|
||||
|
||||
private static long[] computeWaitRead(int[] multipleMessage) {
|
||||
@ -100,9 +110,27 @@ public class TrafficShapingTest extends AbstractSocketTest {
|
||||
return minimalWaitBetween;
|
||||
}
|
||||
|
||||
private static long[] computeWaitAutoRead(int []autoRead) {
|
||||
long [] minimalWaitBetween = new long[autoRead.length + 1];
|
||||
minimalWaitBetween[0] = 0;
|
||||
for (int i = 0; i < autoRead.length; i++) {
|
||||
if (autoRead[i] != 0) {
|
||||
if (autoRead[i] > 0) {
|
||||
minimalWaitBetween[i + 1] = -1;
|
||||
} else {
|
||||
minimalWaitBetween[i + 1] = check;
|
||||
}
|
||||
} else {
|
||||
minimalWaitBetween[i + 1] = 0;
|
||||
}
|
||||
}
|
||||
return minimalWaitBetween;
|
||||
}
|
||||
|
||||
@Test(timeout = 10000)
|
||||
public void testNoTrafficShapping() throws Throwable {
|
||||
logger.info("TEST NO TRAFFIC");
|
||||
TESTNAME = "TEST NO TRAFFIC";
|
||||
TESTRUN = 0;
|
||||
run();
|
||||
}
|
||||
|
||||
@ -113,35 +141,24 @@ public class TrafficShapingTest extends AbstractSocketTest {
|
||||
testTrafficShapping0(sb, cb, false, false, false, false, autoRead, minimalWaitBetween, multipleMessage);
|
||||
}
|
||||
|
||||
@Test(timeout = 15000)
|
||||
public void testExecNoTrafficShapping() throws Throwable {
|
||||
logger.info("TEST EXEC NO TRAFFIC");
|
||||
run();
|
||||
}
|
||||
|
||||
public void testExecNoTrafficShapping(ServerBootstrap sb, Bootstrap cb) throws Throwable {
|
||||
int[] autoRead = null;
|
||||
int[] multipleMessage = { 1, 2, 1 };
|
||||
long[] minimalWaitBetween = null;
|
||||
testTrafficShapping0(sb, cb, true, false, false, false, autoRead, minimalWaitBetween, multipleMessage);
|
||||
}
|
||||
|
||||
@Test(timeout = 10000)
|
||||
public void testWriteTrafficShapping() throws Throwable {
|
||||
logger.info("TEST WRITE");
|
||||
TESTNAME = "TEST WRITE";
|
||||
TESTRUN = 0;
|
||||
run();
|
||||
}
|
||||
|
||||
public void testWriteTrafficShapping(ServerBootstrap sb, Bootstrap cb) throws Throwable {
|
||||
int[] autoRead = null;
|
||||
int[] multipleMessage = { 1, 2, 1 };
|
||||
int[] multipleMessage = { 1, 2, 1, 1 };
|
||||
long[] minimalWaitBetween = computeWaitWrite(multipleMessage);
|
||||
testTrafficShapping0(sb, cb, false, false, true, false, autoRead, minimalWaitBetween, multipleMessage);
|
||||
}
|
||||
|
||||
@Test(timeout = 10000)
|
||||
public void testReadTrafficShapping() throws Throwable {
|
||||
logger.info("TEST READ");
|
||||
TESTNAME = "TEST READ";
|
||||
TESTRUN = 0;
|
||||
run();
|
||||
}
|
||||
|
||||
@ -154,7 +171,8 @@ public class TrafficShapingTest extends AbstractSocketTest {
|
||||
|
||||
@Test(timeout = 10000)
|
||||
public void testWrite1TrafficShapping() throws Throwable {
|
||||
logger.info("TEST WRITE");
|
||||
TESTNAME = "TEST WRITE";
|
||||
TESTRUN = 0;
|
||||
run();
|
||||
}
|
||||
|
||||
@ -167,7 +185,8 @@ public class TrafficShapingTest extends AbstractSocketTest {
|
||||
|
||||
@Test(timeout = 10000)
|
||||
public void testRead1TrafficShapping() throws Throwable {
|
||||
logger.info("TEST READ");
|
||||
TESTNAME = "TEST READ";
|
||||
TESTRUN = 0;
|
||||
run();
|
||||
}
|
||||
|
||||
@ -178,48 +197,24 @@ public class TrafficShapingTest extends AbstractSocketTest {
|
||||
testTrafficShapping0(sb, cb, false, true, false, false, autoRead, minimalWaitBetween, multipleMessage);
|
||||
}
|
||||
|
||||
@Test(timeout = 15000)
|
||||
public void testExecWriteTrafficShapping() throws Throwable {
|
||||
logger.info("TEST EXEC WRITE");
|
||||
run();
|
||||
}
|
||||
|
||||
public void testExecWriteTrafficShapping(ServerBootstrap sb, Bootstrap cb) throws Throwable {
|
||||
int[] autoRead = null;
|
||||
int[] multipleMessage = { 1, 2, 1 };
|
||||
long[] minimalWaitBetween = computeWaitWrite(multipleMessage);
|
||||
testTrafficShapping0(sb, cb, true, false, true, false, autoRead, minimalWaitBetween, multipleMessage);
|
||||
}
|
||||
|
||||
@Test(timeout = 15000)
|
||||
public void testExecReadTrafficShapping() throws Throwable {
|
||||
logger.info("TEST EXEC READ");
|
||||
run();
|
||||
}
|
||||
|
||||
public void testExecReadTrafficShapping(ServerBootstrap sb, Bootstrap cb) throws Throwable {
|
||||
int[] autoRead = null;
|
||||
int[] multipleMessage = { 1, 2, 1, 1 };
|
||||
long[] minimalWaitBetween = computeWaitRead(multipleMessage);
|
||||
testTrafficShapping0(sb, cb, true, true, false, false, autoRead, minimalWaitBetween, multipleMessage);
|
||||
}
|
||||
|
||||
@Test(timeout = 10000)
|
||||
public void testWriteGlobalTrafficShapping() throws Throwable {
|
||||
logger.info("TEST GLOBAL WRITE");
|
||||
TESTNAME = "TEST GLOBAL WRITE";
|
||||
TESTRUN = 0;
|
||||
run();
|
||||
}
|
||||
|
||||
public void testWriteGlobalTrafficShapping(ServerBootstrap sb, Bootstrap cb) throws Throwable {
|
||||
int[] autoRead = null;
|
||||
int[] multipleMessage = { 1, 2, 1 };
|
||||
int[] multipleMessage = { 1, 2, 1, 1 };
|
||||
long[] minimalWaitBetween = computeWaitWrite(multipleMessage);
|
||||
testTrafficShapping0(sb, cb, false, false, true, true, autoRead, minimalWaitBetween, multipleMessage);
|
||||
}
|
||||
|
||||
@Test(timeout = 10000)
|
||||
public void testReadGlobalTrafficShapping() throws Throwable {
|
||||
logger.info("TEST GLOBAL READ");
|
||||
TESTNAME = "TEST GLOBAL READ";
|
||||
TESTRUN = 0;
|
||||
run();
|
||||
}
|
||||
|
||||
@ -232,52 +227,32 @@ public class TrafficShapingTest extends AbstractSocketTest {
|
||||
|
||||
@Test(timeout = 10000)
|
||||
public void testAutoReadTrafficShapping() throws Throwable {
|
||||
logger.info("TEST AUTO READ");
|
||||
TESTNAME = "TEST AUTO READ";
|
||||
TESTRUN = 0;
|
||||
run();
|
||||
}
|
||||
|
||||
public void testAutoReadTrafficShapping(ServerBootstrap sb, Bootstrap cb) throws Throwable {
|
||||
int[] autoRead = { 1, -1, -1, 1, -2, 0, 1, 0, -3, 0, 1, 2, 0 };
|
||||
int[] multipleMessage = { 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1 };
|
||||
long[] minimalWaitBetween = computeWaitRead(multipleMessage);
|
||||
int[] multipleMessage = new int[autoRead.length];
|
||||
Arrays.fill(multipleMessage, 1);
|
||||
long[] minimalWaitBetween = computeWaitAutoRead(autoRead);
|
||||
testTrafficShapping0(sb, cb, false, true, false, false, autoRead, minimalWaitBetween, multipleMessage);
|
||||
}
|
||||
@Test(timeout = 10000)
|
||||
public void testAutoReadGlobalTrafficShapping() throws Throwable {
|
||||
logger.info("TEST AUTO READ GLOBAL");
|
||||
TESTNAME = "TEST AUTO READ GLOBAL";
|
||||
TESTRUN = 0;
|
||||
run();
|
||||
}
|
||||
|
||||
public void testAutoReadGlobalTrafficShapping(ServerBootstrap sb, Bootstrap cb) throws Throwable {
|
||||
int[] autoRead = { 1, -1, -1, 1, -2, 0, 1, 0, -3, 0, 1, 2, 0 };
|
||||
int[] multipleMessage = { 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1 };
|
||||
long[] minimalWaitBetween = computeWaitRead(multipleMessage);
|
||||
int[] multipleMessage = new int[autoRead.length];
|
||||
Arrays.fill(multipleMessage, 1);
|
||||
long[] minimalWaitBetween = computeWaitAutoRead(autoRead);
|
||||
testTrafficShapping0(sb, cb, false, true, false, true, autoRead, minimalWaitBetween, multipleMessage);
|
||||
}
|
||||
@Test(timeout = 10000)
|
||||
public void testAutoReadExecTrafficShapping() throws Throwable {
|
||||
logger.info("TEST AUTO READ EXEC");
|
||||
run();
|
||||
}
|
||||
|
||||
public void testAutoReadExecTrafficShapping(ServerBootstrap sb, Bootstrap cb) throws Throwable {
|
||||
int[] autoRead = { 1, -1, -1, 1, -2, 0, 1, 0, -3, 0, 1, 2, 0 };
|
||||
int[] multipleMessage = { 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1 };
|
||||
long[] minimalWaitBetween = computeWaitRead(multipleMessage);
|
||||
testTrafficShapping0(sb, cb, true, true, false, false, autoRead, minimalWaitBetween, multipleMessage);
|
||||
}
|
||||
@Test(timeout = 10000)
|
||||
public void testAutoReadExecGlobalTrafficShapping() throws Throwable {
|
||||
logger.info("TEST AUTO READ EXEC GLOBAL");
|
||||
run();
|
||||
}
|
||||
|
||||
public void testAutoReadExecGlobalTrafficShapping(ServerBootstrap sb, Bootstrap cb) throws Throwable {
|
||||
int[] autoRead = { 1, -1, -1, 1, -2, 0, 1, 0, -3, 0, 1, 2, 0 };
|
||||
int[] multipleMessage = { 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1 };
|
||||
long[] minimalWaitBetween = computeWaitRead(multipleMessage);
|
||||
testTrafficShapping0(sb, cb, true, true, false, true, autoRead, minimalWaitBetween, multipleMessage);
|
||||
}
|
||||
|
||||
/**
|
||||
*
|
||||
@ -301,25 +276,27 @@ public class TrafficShapingTest extends AbstractSocketTest {
|
||||
* @throws Throwable
|
||||
*/
|
||||
private static void testTrafficShapping0(ServerBootstrap sb, Bootstrap cb, final boolean additionalExecutor,
|
||||
final boolean limitRead, final boolean limitWrite, final boolean globalLimit, int[] autoRead,
|
||||
long[] minimalWaitBetween, int[] multipleMessage) throws Throwable {
|
||||
logger.info("Exec: " + additionalExecutor + " Read: " + limitRead + " Write: " + limitWrite + " Global: "
|
||||
+ globalLimit);
|
||||
final ValidTimestampedHandler sh = new ValidTimestampedHandler(autoRead, multipleMessage);
|
||||
final boolean limitRead, final boolean limitWrite, final boolean globalLimit, int[] autoRead,
|
||||
long[] minimalWaitBetween, int[] multipleMessage) throws Throwable {
|
||||
TESTRUN ++;
|
||||
logger.info("TEST: " + TESTNAME + " RUN: " + TESTRUN +
|
||||
" Exec: " + additionalExecutor + " Read: " + limitRead + " Write: " + limitWrite + " Global: "
|
||||
+ globalLimit);
|
||||
final ServerHandler sh = new ServerHandler(autoRead, multipleMessage);
|
||||
Promise<Boolean> promise = group.next().newPromise();
|
||||
final ClientTrafficHandler ch = new ClientTrafficHandler(promise, minimalWaitBetween, multipleMessage,
|
||||
autoRead);
|
||||
final ClientHandler ch = new ClientHandler(promise, minimalWaitBetween, multipleMessage,
|
||||
autoRead);
|
||||
|
||||
final AbstractTrafficShapingHandler handler;
|
||||
if (limitRead) {
|
||||
if (globalLimit) {
|
||||
handler = new GlobalTrafficShapingHandler(group, 0, bandwidthFactor * messageSize, check);
|
||||
handler = new GlobalTrafficShapingHandler(groupForGlobal, 0, bandwidthFactor * messageSize, check);
|
||||
} else {
|
||||
handler = new ChannelTrafficShapingHandler(0, bandwidthFactor * messageSize, check);
|
||||
}
|
||||
} else if (limitWrite) {
|
||||
if (globalLimit) {
|
||||
handler = new GlobalTrafficShapingHandler(group, bandwidthFactor * messageSize, 0, check);
|
||||
handler = new GlobalTrafficShapingHandler(groupForGlobal, bandwidthFactor * messageSize, 0, check);
|
||||
} else {
|
||||
handler = new ChannelTrafficShapingHandler(bandwidthFactor * messageSize, 0, check);
|
||||
}
|
||||
@ -331,34 +308,18 @@ public class TrafficShapingTest extends AbstractSocketTest {
|
||||
@Override
|
||||
protected void initChannel(SocketChannel c) throws Exception {
|
||||
if (limitRead) {
|
||||
if (additionalExecutor) {
|
||||
c.pipeline().addLast(group, handler);
|
||||
} else {
|
||||
c.pipeline().addLast(handler);
|
||||
}
|
||||
}
|
||||
if (additionalExecutor) {
|
||||
c.pipeline().addLast(group, sh);
|
||||
} else {
|
||||
c.pipeline().addLast(sh);
|
||||
c.pipeline().addLast(TRAFFIC, handler);
|
||||
}
|
||||
c.pipeline().addLast(sh);
|
||||
}
|
||||
});
|
||||
cb.handler(new ChannelInitializer<SocketChannel>() {
|
||||
@Override
|
||||
protected void initChannel(SocketChannel c) throws Exception {
|
||||
if (limitWrite) {
|
||||
if (additionalExecutor) {
|
||||
c.pipeline().addLast(group, handler);
|
||||
} else {
|
||||
c.pipeline().addLast(handler);
|
||||
}
|
||||
}
|
||||
if (additionalExecutor) {
|
||||
c.pipeline().addLast(group, ch);
|
||||
} else {
|
||||
c.pipeline().addLast(ch);
|
||||
c.pipeline().addLast(TRAFFIC, handler);
|
||||
}
|
||||
c.pipeline().addLast(ch);
|
||||
}
|
||||
});
|
||||
|
||||
@ -381,7 +342,8 @@ public class TrafficShapingTest extends AbstractSocketTest {
|
||||
assertTrue("Error during exceution of TrafficShapping: " + promise.cause(), promise.isSuccess());
|
||||
|
||||
float average = (totalNb * messageSize) / (float) (stop - start);
|
||||
logger.info("Average of traffic: " + average + " compare to " + bandwidthFactor);
|
||||
logger.info("TEST: " + TESTNAME + " RUN: " + TESTRUN +
|
||||
" Average of traffic: " + average + " compare to " + bandwidthFactor);
|
||||
sh.channel.close().sync();
|
||||
ch.channel.close().sync();
|
||||
sc.close().sync();
|
||||
@ -392,13 +354,13 @@ public class TrafficShapingTest extends AbstractSocketTest {
|
||||
|
||||
if (autoRead == null && minimalWaitBetween != null) {
|
||||
assertTrue("Overall Traffic not ok since > " + maxfactor + ": " + average,
|
||||
average <= maxfactor);
|
||||
average <= maxfactor);
|
||||
if (additionalExecutor) {
|
||||
// Oio is not as good when using additionalExecutor
|
||||
assertTrue("Overall Traffic not ok since < 0.25: " + average, average >= 0.25);
|
||||
} else {
|
||||
assertTrue("Overall Traffic not ok since < " + minfactor + ": " + average,
|
||||
average >= minfactor);
|
||||
average >= minfactor);
|
||||
}
|
||||
}
|
||||
if (handler != null && globalLimit) {
|
||||
@ -419,7 +381,7 @@ public class TrafficShapingTest extends AbstractSocketTest {
|
||||
}
|
||||
}
|
||||
|
||||
private static class ClientTrafficHandler extends SimpleChannelInboundHandler<ByteBuf> {
|
||||
private static class ClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
|
||||
volatile Channel channel;
|
||||
final AtomicReference<Throwable> exception = new AtomicReference<Throwable>();
|
||||
volatile int step;
|
||||
@ -430,8 +392,8 @@ public class TrafficShapingTest extends AbstractSocketTest {
|
||||
private final int[] autoRead;
|
||||
final Promise<Boolean> promise;
|
||||
|
||||
ClientTrafficHandler(Promise<Boolean> promise, long[] minimalWaitBetween, int[] multipleMessage,
|
||||
int[] autoRead) {
|
||||
ClientHandler(Promise<Boolean> promise, long[] minimalWaitBetween, int[] multipleMessage,
|
||||
int[] autoRead) {
|
||||
this.minimalWaitBetween = minimalWaitBetween;
|
||||
this.multipleMessage = Arrays.copyOf(multipleMessage, multipleMessage.length);
|
||||
this.promise = promise;
|
||||
@ -444,8 +406,9 @@ public class TrafficShapingTest extends AbstractSocketTest {
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void messageReceived(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
|
||||
public void messageReceived(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
|
||||
long lastTimestamp = 0;
|
||||
loggerClient.debug("Step: " + step + " Read: " + (in.readableBytes() / 8) + " blocks");
|
||||
while (in.isReadable()) {
|
||||
lastTimestamp = in.readLong();
|
||||
multipleMessage[step]--;
|
||||
@ -459,20 +422,12 @@ public class TrafficShapingTest extends AbstractSocketTest {
|
||||
if (autoRead != null) {
|
||||
if (step > 0 && autoRead[step - 1] != 0) {
|
||||
ar = autoRead[step - 1];
|
||||
if (ar > 0) {
|
||||
minimalWait = -1;
|
||||
} else {
|
||||
minimalWait = minimalms / 3;
|
||||
}
|
||||
} else {
|
||||
minimalWait = 0;
|
||||
}
|
||||
}
|
||||
logger.info("Step: " + step + " Interval: " + (lastTimestamp - currentLastTime) + " compareTo "
|
||||
+ minimalWait + " (" + ar + ")");
|
||||
/* Comment since it seems not enough precise/correct in Master
|
||||
loggerClient.info("Step: " + step + " Interval: " + (lastTimestamp - currentLastTime) + " compareTo "
|
||||
+ minimalWait + " (" + ar + ")");
|
||||
assertTrue("The interval of time is incorrect:" + (lastTimestamp - currentLastTime) + " not> "
|
||||
+ minimalWait, lastTimestamp - currentLastTime >= minimalWait);*/
|
||||
+ minimalWait, lastTimestamp - currentLastTime >= minimalWait);
|
||||
currentLastTime = lastTimestamp;
|
||||
step++;
|
||||
if (multipleMessage.length > step) {
|
||||
@ -496,14 +451,14 @@ public class TrafficShapingTest extends AbstractSocketTest {
|
||||
}
|
||||
}
|
||||
|
||||
private static class ValidTimestampedHandler extends SimpleChannelInboundHandler<ByteBuf> {
|
||||
private static class ServerHandler extends SimpleChannelInboundHandler<ByteBuf> {
|
||||
private final int[] autoRead;
|
||||
private final int[] multipleMessage;
|
||||
volatile Channel channel;
|
||||
volatile int step;
|
||||
final AtomicReference<Throwable> exception = new AtomicReference<Throwable>();
|
||||
|
||||
ValidTimestampedHandler(int[] autoRead, int[] multipleMessage) {
|
||||
ServerHandler(int[] autoRead, int[] multipleMessage) {
|
||||
this.autoRead = autoRead;
|
||||
this.multipleMessage = Arrays.copyOf(multipleMessage, multipleMessage.length);
|
||||
}
|
||||
@ -514,57 +469,66 @@ public class TrafficShapingTest extends AbstractSocketTest {
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void messageReceived(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
|
||||
public void messageReceived(final ChannelHandlerContext ctx, ByteBuf in) throws Exception {
|
||||
byte[] actual = new byte[in.readableBytes()];
|
||||
int nb = actual.length / messageSize;
|
||||
loggerServer.info("Step: " + step + " Read: " + nb + " blocks");
|
||||
in.readBytes(actual);
|
||||
long timestamp = System.currentTimeMillis();
|
||||
int nb = actual.length / messageSize;
|
||||
int isAutoRead = 0;
|
||||
int laststep = step;
|
||||
for (int i = 0; i < nb; i++) {
|
||||
multipleMessage[step]--;
|
||||
if (multipleMessage[step] == 0) {
|
||||
// setAutoRead test
|
||||
if (autoRead != null) {
|
||||
isAutoRead = autoRead[step];
|
||||
}
|
||||
step++;
|
||||
}
|
||||
}
|
||||
if (laststep != step && autoRead != null && isAutoRead != 2) {
|
||||
if (isAutoRead != 0) {
|
||||
logger.info("Set AutoRead: " + (isAutoRead > 0) + " Step: " + step);
|
||||
channel.config().setAutoRead(isAutoRead > 0);
|
||||
} else {
|
||||
logger.info("AutoRead: NO Step:" + step);
|
||||
if (laststep != step) {
|
||||
// setAutoRead test
|
||||
if (autoRead != null && isAutoRead != 2) {
|
||||
if (isAutoRead != 0) {
|
||||
loggerServer.info("Step: " + step + " Set AutoRead: " + (isAutoRead > 0));
|
||||
channel.config().setAutoRead(isAutoRead > 0);
|
||||
} else {
|
||||
loggerServer.info("Step: " + step + " AutoRead: NO");
|
||||
}
|
||||
}
|
||||
}
|
||||
logger.debug("Get: " + actual.length + " TS " + timestamp + " NB: " + nb);
|
||||
loggerServer.debug("Step: " + step + " Write: " + nb);
|
||||
for (int i = 0; i < nb; i++) {
|
||||
channel.write(Unpooled.copyLong(timestamp));
|
||||
}
|
||||
channel.flush();
|
||||
if (laststep != step && isAutoRead != 0) {
|
||||
if (isAutoRead < 0) {
|
||||
final int exactStep = step;
|
||||
long wait = (isAutoRead == -1) ? minimalms : stepms + minimalms;
|
||||
if (isAutoRead == -3) {
|
||||
wait = stepms * 3;
|
||||
}
|
||||
executor.schedule(new Runnable() {
|
||||
public void run() {
|
||||
logger.info("Reset AutoRead: Step " + exactStep);
|
||||
channel.config().setAutoRead(true);
|
||||
if (laststep != step) {
|
||||
// setAutoRead test
|
||||
if (isAutoRead != 0) {
|
||||
if (isAutoRead < 0) {
|
||||
final int exactStep = step;
|
||||
long wait = (isAutoRead == -1) ? minimalms : stepms + minimalms;
|
||||
if (isAutoRead == -3) {
|
||||
wait = stepms * 3;
|
||||
}
|
||||
}, wait, TimeUnit.MILLISECONDS);
|
||||
} else {
|
||||
if (isAutoRead > 1) {
|
||||
logger.info("Will Set AutoRead: Rrue, Step: " + step);
|
||||
executor.schedule(new Runnable() {
|
||||
public void run() {
|
||||
logger.info("Set AutoRead: Rrue, Step: " + step);
|
||||
loggerServer.info("Step: " + exactStep + " Reset AutoRead");
|
||||
channel.config().setAutoRead(true);
|
||||
}
|
||||
}, stepms + minimalms, TimeUnit.MILLISECONDS);
|
||||
}, wait, TimeUnit.MILLISECONDS);
|
||||
} else {
|
||||
if (isAutoRead > 1) {
|
||||
loggerServer.debug("Step: " + step + " Will Set AutoRead: True");
|
||||
final int exactStep = step;
|
||||
executor.schedule(new Runnable() {
|
||||
public void run() {
|
||||
loggerServer.info("Step: " + exactStep + " Set AutoRead: True");
|
||||
channel.config().setAutoRead(true);
|
||||
}
|
||||
}, stepms + minimalms, TimeUnit.MILLISECONDS);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user