[GitHub] merlimat closed pull request #3172: Modify producer handler of WebSocket to send ack to client asynchronously

[GitBox <gi...@apache.org>]

merlimat closed pull request #3172: Modify producer handler of WebSocket to send ack to client asynchronously 
URL: https://github.com/apache/pulsar/pull/3172
 
 
   

This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:

As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):

diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ProducerHandler.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ProducerHandler.java
index bca9762c2f..338c0f1ebd 100644
--- a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ProducerHandler.java
+++ b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ProducerHandler.java
@@ -52,6 +52,7 @@
 import org.apache.pulsar.websocket.data.ProducerAck;
 import org.apache.pulsar.websocket.data.ProducerMessage;
 import org.apache.pulsar.websocket.stats.StatsBuckets;
+import org.eclipse.jetty.websocket.api.WriteCallback;
 import org.eclipse.jetty.websocket.servlet.ServletUpgradeResponse;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -243,7 +244,20 @@ protected Boolean isAuthorized(String authRole, AuthenticationDataSource authent
     private void sendAckResponse(ProducerAck response) {
         try {
             String msg = ObjectMapperFactory.getThreadLocal().writeValueAsString(response);
-            getSession().getRemote().sendString(msg);
+            getSession().getRemote().sendString(msg, new WriteCallback() {
+                @Override
+                public void writeFailed(Throwable th) {
+                    log.warn("[{}] Failed to send ack {}", producer.getTopic(), th.getMessage(), th);
+                }
+
+                @Override
+                public void writeSuccess() {
+                    if (log.isDebugEnabled()) {
+                        log.debug("[{}] Ack was sent successfully to {}", producer.getTopic(),
+                                getRemote().getInetSocketAddress().toString());
+                    }
+                }
+            });
         } catch (JsonProcessingException e) {
             log.warn("[{}] Failed to generate ack json-response {}", producer.getTopic(), e.getMessage(), e);
         } catch (Exception e) {


 

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services