@@ -68,9 +68,10 @@ public static void main(String[] args) {
68
68
String hostName = optArg (args , 0 , "localhost" );
69
69
int portNumber = optArg (args , 1 , AMQP .PROTOCOL .PORT );
70
70
boolean writeStats = optArg (args , 2 , true );
71
+ boolean noAck = optArg (args , 3 , true );
71
72
final Connection conn = new ConnectionFactory ().newConnection (hostName , portNumber );
72
73
System .out .println ("Channel 0 fully open." );
73
- new ConsumerMain (conn , writeStats ).run ();
74
+ new ConsumerMain (conn , writeStats , noAck ).run ();
74
75
} catch (Exception e ) {
75
76
System .err .println ("Main thread caught exception: " + e );
76
77
e .printStackTrace ();
@@ -90,10 +91,14 @@ public static void sleep(int ms) {
90
91
91
92
public boolean _writeStats ;
92
93
93
- public ConsumerMain (Connection connection , boolean writeStats ) {
94
+ public boolean _noAck ;
95
+
96
+ public ConsumerMain (Connection connection , boolean writeStats , boolean noAck ) {
94
97
_connection = connection ;
95
98
_writeStats = writeStats ;
99
+ _noAck = noAck ;
96
100
System .out .println ((_writeStats ? "WILL" : "WON'T" ) + " write statistics." );
101
+ System .out .println ((_noAck ? "WILL" : "WON'T" ) + " use server-side auto-acking." );
97
102
}
98
103
99
104
public void run () {
@@ -119,8 +124,9 @@ private void runIt() throws IOException {
119
124
channel .queueBind (completionQueue , exchangeName , "" );
120
125
121
126
LatencyExperimentConsumer callback = new LatencyExperimentConsumer (channel , queueName );
127
+ callback ._noAck = this ._noAck ;
122
128
123
- channel .basicConsume (queueName , true , callback );
129
+ channel .basicConsume (queueName , _noAck , callback );
124
130
channel .basicConsume (completionQueue , true , "completion" , callback );
125
131
callback .report (_writeStats );
126
132
0 commit comments