optimize indention in ServerCnx#handleProducer
### Motivation
Too many indentions in ServerCnx#handleProducer, not good for readability and further developing.
### Modifications
Move else part of isAuthorized up front.
Some other minor code style update, no code logical changed.
@@ -1143,14 +1143,18 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
topicName, TopicOperation.PRODUCE
);
isAuthorizedFuture.thenApply(isAuthorized -> {
- if (isAuthorized) {
+ if (!isAuthorized) {
+ String msg = "Client is not authorized to Produce";
+ log.warn("[{}] {} with role {}", remoteAddress, msg, getPrincipal());
+ ctx.writeAndFlush(Commands.newError(requestId, ServerError.AuthorizationError, msg));
+ return null;
+ }
+
if (log.isDebugEnabled()) {
- log.debug("[{}] Client is authorized to Produce with role {}",
- remoteAddress, getPrincipal());
+ log.debug("[{}] Client is authorized to Produce with role {}", remoteAddress, getPrincipal());
}
CompletableFuture<Producer> producerFuture = new CompletableFuture<>();
- CompletableFuture<Producer> existingProducerFuture = producers.putIfAbsent(producerId,
- producerFuture);
+ CompletableFuture<Producer> existingProducerFuture = producers.putIfAbsent(producerId, producerFuture);
if (existingProducerFuture != null) {
if (existingProducerFuture.isDone() && !existingProducerFuture.isCompletedExceptionally()) {
@@ -1159,7 +1163,6 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
+ " producerId={}, producer={}", remoteAddress, producerId, producer);
commandSender.sendProducerSuccessResponse(requestId, producer.getProducerName(),
producer.getSchemaVersion());
-
return null;
} else {
// There was an early request to create a producer with same producerId.
@@ -1174,11 +1177,9 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
// remove producer with producerId as it's already completed with exception
producers.remove(producerId, existingProducerFuture);
}
- log.warn("[{}][{}] Producer with id is already present on the connection,"
- + " producerId={}", remoteAddress, topicName, producerId);
- commandSender.sendErrorResponse(requestId, error,
- "Producer is already present on the connection");
-
+ log.warn("[{}][{}] Producer with id is already present on the connection, producerId={}",
+ remoteAddress, topicName, producerId);
+ commandSender.sendErrorResponse(requestId, error, "Producer is already present on the connection");
return null;
}
}
@@ -1188,8 +1189,7 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
service.getOrCreateTopic(topicName.toString()).thenAccept((Topic topic) -> {
// Before creating producer, check if backlog quota exceeded
// on topic for size based limit and time based limit
- for (BacklogQuota.BacklogQuotaType backlogQuotaType :
- BacklogQuota.BacklogQuotaType.values()) {
+ for (BacklogQuota.BacklogQuotaType backlogQuotaType : BacklogQuota.BacklogQuotaType.values()) {
if (topic.isBacklogQuotaExceeded(producerName, backlogQuotaType)) {
IllegalStateException illegalStateException = new IllegalStateException(
"Cannot create producer on topic with backlog quota exceeded");
@@ -1209,7 +1209,6 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
return;
}
}
-
// Check whether the producer will publish encrypted messages or not
if ((topic.isEncryptionRequired() || encryptionRequireOnProducer) && !isEncrypted) {
String msg = String.format("Encryption is required in %s", topicName);
@@ -1270,14 +1269,8 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
BrokerServiceException.getClientErrorCode(cause), cause.getMessage());
}
producers.remove(producerId, producerFuture);
-
return null;
});
- } else {
- String msg = "Client is not authorized to Produce";
- log.warn("[{}] {} with role {}", remoteAddress, msg, getPrincipal());
- ctx.writeAndFlush(Commands.newError(requestId, ServerError.AuthorizationError, msg));
- }
return null;
}).exceptionally(ex -> {
logAuthException(remoteAddress, "producer", getPrincipal(), Optional.of(topicName), ex);
