Skip to content

Commit da8b126

Browse files
acogoluegnesacogoluegnes
committed
Sonar fixes
1 parent f613ad5 commit da8b126

File tree

6 files changed

+108
-48
lines changed

6 files changed

+108
-48
lines changed

‎pom.xml‎

Lines changed: 40 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -298,7 +298,7 @@
298298
needed because of bug in OpenJDK 8 u181 on Debian distros
299299
see https://stackoverflow.com/questions/53010200/maven-surefire-could-not-find-forkedbooter-class
300300
-->
301-
<argLine>-Djdk.net.URLClassPath.disableClassPathURLCheck=true</argLine>
301+
<argLine>-Djdk.net.URLClassPath.disableClassPathURLCheck=true ${argLine}</argLine>
302302
</configuration>
303303
</plugin>
304304

@@ -314,7 +314,7 @@
314314
needed because of bug in OpenJDK 8 u181 on Debian distros
315315
see https://stackoverflow.com/questions/53010200/maven-surefire-could-not-find-forkedbooter-class
316316
-->
317-
<argLine>-Djdk.net.URLClassPath.disableClassPathURLCheck=true</argLine>
317+
<argLine>-Djdk.net.URLClassPath.disableClassPathURLCheck=true ${argLine}</argLine>
318318
</configuration>
319319

320320
<executions>
@@ -327,6 +327,44 @@
327327
</executions>
328328
</plugin>
329329

330+
<plugin>
331+
<groupId>org.jacoco</groupId>
332+
<artifactId>jacoco-maven-plugin</artifactId>
333+
<version>0.8.2</version>
334+
<executions>
335+
<execution>
336+
<id>default-prepare-agent</id>
337+
<goals>
338+
<goal>prepare-agent</goal>
339+
</goals>
340+
</execution>
341+
<execution>
342+
<id>default-prepare-agent-integration</id>
343+
<goals>
344+
<goal>prepare-agent-integration</goal>
345+
</goals>
346+
</execution>
347+
<execution>
348+
<id>default-report</id>
349+
<goals>
350+
<goal>report</goal>
351+
</goals>
352+
</execution>
353+
<execution>
354+
<id>default-report-integration</id>
355+
<goals>
356+
<goal>report-integration</goal>
357+
</goals>
358+
</execution>
359+
<execution>
360+
<id>default-check</id>
361+
<goals>
362+
<goal>check</goal>
363+
</goals>
364+
</execution>
365+
</executions>
366+
</plugin>
367+
330368
<plugin>
331369
<groupId>org.codehaus.mojo</groupId>
332370
<artifactId>versions-maven-plugin</artifactId>

‎src/main/java/com/rabbitmq/perf/BenchmarkResults.java‎

Lines changed: 2 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -42,19 +42,13 @@ public static void main(String[] args) throws Exception {
4242
System.out.println("Usage: BenchmarkResults result-json-file");
4343
System.exit(1);
4444
}
45-
BufferedReader reader = null;
4645
StringBuilder builder = new StringBuilder();
47-
try {
48-
File resultsFile = new File(args[0]);
49-
reader = new BufferedReader(new FileReader(resultsFile));
46+
File resultsFile = new File(args[0]);
47+
try (BufferedReader reader = new BufferedReader(new FileReader(resultsFile))){
5048
String line;
5149
while ((line = reader.readLine()) != null) {
5250
builder.append(line);
5351
}
54-
} finally {
55-
if(reader != null) {
56-
reader.close();
57-
}
5852
}
5953

6054
System.setProperty("org.eclipse.jetty.LEVEL", "WARN");

‎src/main/java/com/rabbitmq/perf/Consumer.java‎

Lines changed: 20 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -137,19 +137,8 @@ public void handleDelivery(String consumerTag, Envelope envelope, BasicPropertie
137137
long messageTimestamp = timestampExtractor.apply(properties, body);
138138
long nowTimestamp = timestampProvider.getCurrentTime();
139139

140-
if (!autoAck) {
141-
dealWithWriteOperation(() -> {
142-
if (multiAckEvery == 0) {
143-
channel.basicAck(envelope.getDeliveryTag(), false);
144-
} else if (currentMessageCount % multiAckEvery == 0) {
145-
channel.basicAck(envelope.getDeliveryTag(), true);
146-
}
147-
}, recoveryProcess);
148-
}
149-
150-
if (txSize != 0 && currentMessageCount % txSize == 0) {
151-
dealWithWriteOperation(() -> channel.txCommit(), recoveryProcess);
152-
}
140+
ackIfNecessary(envelope, currentMessageCount);
141+
commitTransactionIfNecessary(currentMessageCount);
153142

154143
long diff_time = timestampProvider.getDifference(nowTimestamp, messageTimestamp);
155144
stats.handleRecv(id.equals(envelope.getRoutingKey()) ? diff_time : 0L);
@@ -165,6 +154,24 @@ public void handleDelivery(String consumerTag, Envelope envelope, BasicPropertie
165154
}
166155
}
167156

157+
private void ackIfNecessary(Envelope envelope, int currentMessageCount) throws IOException {
158+
if (!autoAck) {
159+
dealWithWriteOperation(() -> {
160+
if (multiAckEvery == 0) {
161+
channel.basicAck(envelope.getDeliveryTag(), false);
162+
} else if (currentMessageCount % multiAckEvery == 0) {
163+
channel.basicAck(envelope.getDeliveryTag(), true);
164+
}
165+
}, recoveryProcess);
166+
}
167+
}
168+
169+
private void commitTransactionIfNecessary(int currentMessageCount) throws IOException {
170+
if (txSize != 0 && currentMessageCount % txSize == 0) {
171+
dealWithWriteOperation(() -> channel.txCommit(), recoveryProcess);
172+
}
173+
}
174+
168175
@Override
169176
public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {
170177
LOGGER.debug(

‎src/main/java/com/rabbitmq/perf/LocalFilesMessageBodySource.java‎

Lines changed: 1 addition & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -37,18 +37,11 @@ public LocalFilesMessageBodySource(List<String> filesNames, String contentType)
3737
if (!file.exists() || file.isDirectory()) {
3838
throw new IllegalArgumentException(fileName + " isn't a valid body file.");
3939
}
40-
BufferedInputStream inputStream = null;
41-
try {
42-
inputStream = new BufferedInputStream(new FileInputStream(file));
40+
try (BufferedInputStream inputStream = new BufferedInputStream(new FileInputStream(file))) {
4341
byte [] body = new byte[(int) file.length()];
4442
inputStream.read(body, 0, body.length);
4543
bodies.add(body);
46-
} finally {
47-
if (inputStream != null) {
48-
inputStream.close();
49-
}
5044
}
51-
5245
}
5346
this.contentType = contentType;
5447
}

‎src/main/java/com/rabbitmq/perf/MulticastSet.java‎

Lines changed: 44 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -62,6 +62,7 @@ public class MulticastSet {
6262
* There's a command line argument to override this anyway.
6363
*/
6464
private static final int PUBLISHING_INTERVAL_NB_PRODUCERS_PER_THREAD = 50;
65+
private static final String PRODUCER_THREAD_PREFIX = "perf-test-producer-";
6566
private final Stats stats;
6667
private final ConnectionFactory factory;
6768
private final MulticastParams params;
@@ -124,6 +125,32 @@ public void run(boolean announceStartup)
124125

125126
Runnable[] consumerRunnables = new Runnable[params.getConsumerThreadCount()];
126127
Connection[] consumerConnections = new Connection[params.getConsumerCount()];
128+
Function<Integer, ExecutorService> consumersExecutorsFactory;
129+
consumersExecutorsFactory = createConsumersExecutorsFactory();
130+
131+
createConsumers(announceStartup, consumerRunnables, consumerConnections, consumersExecutorsFactory);
132+
133+
this.params.resetTopologyHandler();
134+
135+
AgentState[] producerStates = new AgentState[params.getProducerThreadCount()];
136+
Connection[] producerConnections = new Connection[params.getProducerCount()];
137+
// producers don't need an executor service, as they don't have any consumers
138+
// this consumer should never be asked to create any threads
139+
ExecutorService executorServiceForProducersConsumers = this.threadingHandler.executorService(
140+
"perf-test-producers-worker-", 0
141+
);
142+
factory.setSharedExecutor(executorServiceForProducersConsumers);
143+
createProducers(announceStartup, producerStates, producerConnections);
144+
145+
startConsumers(consumerRunnables);
146+
startProducers(producerStates);
147+
148+
this.completionHandler.waitForCompletion();
149+
150+
shutdown(configurationConnection, consumerConnections, producerStates, producerConnections);
151+
}
152+
153+
private Function<Integer, ExecutorService> createConsumersExecutorsFactory() {
127154
Function<Integer, ExecutorService> consumersExecutorsFactory;
128155
if (params.getConsumersThreadPools() > 0) {
129156
consumersExecutorsFactory = new CacheConsumersExecutorsFactory(
@@ -134,6 +161,10 @@ public void run(boolean announceStartup)
134161
this.threadingHandler, this.params
135162
);
136163
}
164+
return consumersExecutorsFactory;
165+
}
166+
167+
private void createConsumers(boolean announceStartup, Runnable[] consumerRunnables, Connection[] consumerConnections, Function<Integer, ExecutorService> consumersExecutorsFactory) throws URISyntaxException, NoSuchAlgorithmException, KeyManagementException, IOException, TimeoutException {
137168
for (int i = 0; i < consumerConnections.length; i++) {
138169
if (announceStartup) {
139170
System.out.println("id: " + testID + ", starting consumer #" + i);
@@ -151,23 +182,15 @@ public void run(boolean announceStartup)
151182
consumerRunnables[(i * params.getConsumerChannelCount()) + j] = params.createConsumer(consumerConnection, stats, this.completionHandler);
152183
}
153184
}
185+
}
154186

155-
this.params.resetTopologyHandler();
156-
157-
AgentState[] producerStates = new AgentState[params.getProducerThreadCount()];
158-
Connection[] producerConnections = new Connection[params.getProducerCount()];
159-
// producers don't need an executor service, as they don't have any consumers
160-
// this consumer should never be asked to create any threads
161-
ExecutorService executorServiceForProducersConsumers = this.threadingHandler.executorService(
162-
"perf-test-producers-worker-", 0
163-
);
164-
factory.setSharedExecutor(executorServiceForProducersConsumers);
187+
private void createProducers(boolean announceStartup, AgentState[] producerStates, Connection[] producerConnections) throws URISyntaxException, NoSuchAlgorithmException, KeyManagementException, IOException, TimeoutException {
165188
for (int i = 0; i < producerConnections.length; i++) {
166189
if (announceStartup) {
167190
System.out.println("id: " + testID + ", starting producer #" + i);
168191
}
169192
setUri();
170-
Connection producerConnection = factory.newConnection("perf-test-producer-" + i);
193+
Connection producerConnection = factory.newConnection(PRODUCER_THREAD_PREFIX + i);
171194
producerConnections[i] = producerConnection;
172195
for (int j = 0; j < params.getProducerChannelCount(); j++) {
173196
if (announceStartup) {
@@ -178,18 +201,22 @@ public void run(boolean announceStartup)
178201
producerStates[(i * params.getProducerChannelCount()) + j] = agentState;
179202
}
180203
}
204+
}
181205

206+
private void startConsumers(Runnable[] consumerRunnables) throws InterruptedException {
182207
for (Runnable runnable : consumerRunnables) {
183208
runnable.run();
184209
if (params.getConsumerSlowStart()) {
185210
System.out.println("Delaying start by 1 second because -S/--slow-start was requested");
186211
Thread.sleep(1000);
187212
}
188213
}
214+
}
189215

216+
private void startProducers(AgentState[] producerStates) {
190217
if (params.getPublishingInterval() > 0) {
191218
ScheduledExecutorService producersExecutorService = this.threadingHandler.scheduledExecutorService(
192-
"perf-test-producer-", nbThreadsForConsumer(params)
219+
PRODUCER_THREAD_PREFIX, nbThreadsForConsumer(params)
193220
);
194221
Supplier<Integer> startDelaySupplier;
195222
if (params.getProducerRandomStartDelayInSeconds() > 0) {
@@ -203,21 +230,21 @@ public void run(boolean announceStartup)
203230
AgentState producerState = producerStates[i];
204231
int delay = startDelaySupplier.get();
205232
producerState.task = producersExecutorService.scheduleAtFixedRate(
206-
producerState.runnable.createRunnableForScheduling(),
207-
delay, publishingInterval, TimeUnit.SECONDS
233+
producerState.runnable.createRunnableForScheduling(),
234+
delay, publishingInterval, TimeUnit.SECONDS
208235
);
209236
}
210237
} else {
211238
ExecutorService producersExecutorService = this.threadingHandler.executorService(
212-
"perf-test-producer-", producerStates.length
239+
PRODUCER_THREAD_PREFIX, producerStates.length
213240
);
214241
for (AgentState producerState : producerStates) {
215242
producerState.task = producersExecutorService.submit(producerState.runnable);
216243
}
217244
}
245+
}
218246

219-
this.completionHandler.waitForCompletion();
220-
247+
private void shutdown(Connection configurationConnection, Connection[] consumerConnections, AgentState[] producerStates, Connection[] producerConnections) {
221248
try {
222249
LOGGER.debug("Starting test shutdown");
223250
for (AgentState producerState : producerStates) {

‎src/main/java/com/rabbitmq/perf/MulticastValue.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ public void setup(MulticastParams params) {
2929
}
3030

3131
public void teardown(MulticastParams params) {
32+
// nothing to close here
3233
}
3334

3435
public String getName() {

0 commit comments

Comments
 (0)