You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2018/03/24 00:50:32 UTC
[GitHub] merlimat closed pull request #1429: Add v2 support to the websockets endpoints.
merlimat closed pull request #1429: Add v2 support to the websockets endpoints.
URL: https://github.com/apache/incubator-pulsar/pull/1429
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-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
index 51014a7e1..98a4e08e5 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
@@ -80,6 +80,7 @@
import org.apache.pulsar.zookeeper.ZookeeperBkClientFactoryImpl;
import org.apache.zookeeper.ZooKeeper;
import org.eclipse.jetty.servlet.ServletHolder;
+import org.eclipse.jetty.websocket.servlet.WebSocketServlet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -310,12 +311,24 @@ public void start() throws PulsarServerException {
new ClusterData(webServiceAddress, webServiceAddressTls, brokerServiceUrl, brokerServiceUrlTls),
config);
this.webSocketService.start();
+
+ final WebSocketServlet producerWebSocketServlet = new WebSocketProducerServlet(webSocketService);
this.webService.addServlet(WebSocketProducerServlet.SERVLET_PATH,
- new ServletHolder(new WebSocketProducerServlet(webSocketService)), true, attributeMap);
+ new ServletHolder(producerWebSocketServlet), true, attributeMap);
+ this.webService.addServlet(WebSocketProducerServlet.SERVLET_PATH_V2,
+ new ServletHolder(producerWebSocketServlet), true, attributeMap);
+
+ final WebSocketServlet consumerWebSocketServlet = new WebSocketConsumerServlet(webSocketService);
this.webService.addServlet(WebSocketConsumerServlet.SERVLET_PATH,
- new ServletHolder(new WebSocketConsumerServlet(webSocketService)), true, attributeMap);
+ new ServletHolder(consumerWebSocketServlet), true, attributeMap);
+ this.webService.addServlet(WebSocketConsumerServlet.SERVLET_PATH_V2,
+ new ServletHolder(consumerWebSocketServlet), true, attributeMap);
+
+ final WebSocketServlet readerWebSocketServlet = new WebSocketReaderServlet(webSocketService);
this.webService.addServlet(WebSocketReaderServlet.SERVLET_PATH,
- new ServletHolder(new WebSocketReaderServlet(webSocketService)), true, attributeMap);
+ new ServletHolder(readerWebSocketServlet), true, attributeMap);
+ this.webService.addServlet(WebSocketReaderServlet.SERVLET_PATH_V2,
+ new ServletHolder(readerWebSocketServlet), true, attributeMap);
}
if (LOG.isDebugEnabled()) {
diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/AbstractWebSocketHandler.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/AbstractWebSocketHandler.java
index 97f73d245..7e06775ac 100644
--- a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/AbstractWebSocketHandler.java
+++ b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/AbstractWebSocketHandler.java
@@ -32,6 +32,7 @@
import org.apache.pulsar.broker.authentication.AuthenticationDataHttps;
import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
+import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
import org.eclipse.jetty.websocket.api.Session;
import org.eclipse.jetty.websocket.api.WebSocketAdapter;
@@ -46,9 +47,10 @@
protected final WebSocketService service;
protected final HttpServletRequest request;
- protected final String topic;
+ protected final TopicName topic;
protected final Map<String, String> queryParams;
+
public AbstractWebSocketHandler(WebSocketService service, HttpServletRequest request, ServletUpgradeResponse response) {
this.service = service;
this.request = request;
@@ -149,22 +151,39 @@ protected String checkAuthentication() {
return null;
}
- private String extractTopicName(HttpServletRequest request) {
+ private TopicName extractTopicName(HttpServletRequest request) {
String uri = request.getRequestURI();
List<String> parts = Splitter.on("/").splitToList(uri);
- // Format must be like :
+ // V1 Format must be like :
// /ws/producer/persistent/my-property/my-cluster/my-ns/my-topic
// or
// /ws/consumer/persistent/my-property/my-cluster/my-ns/my-topic/my-subscription
// or
// /ws/reader/persistent/my-property/my-cluster/my-ns/my-topic
+
+ // V2 Format must be like :
+ // /ws/v2/producer/persistent/my-property/my-ns/my-topic
+ // or
+ // /ws/v2/consumer/persistent/my-property/my-ns/my-topic/my-subscription
+ // or
+ // /ws/v2/reader/persistent/my-property/my-ns/my-topic
+
checkArgument(parts.size() >= 8, "Invalid topic name format");
checkArgument(parts.get(1).equals("ws"));
- checkArgument(parts.get(3).equals("persistent") || parts.get(3).equals("non-persistent"));
- TopicName topicName = TopicName.get(parts.get(3), parts.get(4), parts.get(5), parts.get(6), parts.get(7));
- return topicName.toString();
+ final boolean isV2Format = parts.get(2).equals("v2");
+ final int domainIndex = isV2Format ? 4 : 3;
+ checkArgument(parts.get(domainIndex).equals("persistent") ||
+ parts.get(domainIndex).equals("non-persistent"));
+
+
+ final String domain = parts.get(domainIndex);
+ final NamespaceName namespace = isV2Format ? NamespaceName.get(parts.get(5), parts.get(6)) :
+ NamespaceName.get( parts.get(4), parts.get(5), parts.get(6));
+ final String name = parts.get(7);
+
+ return TopicName.get(domain, namespace, name);
}
protected abstract Boolean isAuthorized(String authRole, AuthenticationDataSource authenticationData) throws Exception;
diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ConsumerHandler.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ConsumerHandler.java
index 8b0f57307..aea88f343 100644
--- a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ConsumerHandler.java
+++ b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ConsumerHandler.java
@@ -35,10 +35,9 @@
import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.ConsumerConfiguration;
-import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.PulsarClientException.ConsumerBusyException;
+import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.SubscriptionType;
-import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.util.DateFormatter;
import org.apache.pulsar.common.util.ObjectMapperFactory;
import org.apache.pulsar.websocket.data.ConsumerAck;
@@ -91,7 +90,7 @@ public ConsumerHandler(WebSocketService service, HttpServletRequest request, Ser
return;
}
- this.consumer = service.getPulsarClient().subscribe(topic, subscription, conf);
+ this.consumer = service.getPulsarClient().subscribe(topic.toString(), subscription, conf);
if (!this.service.addConsumer(this)) {
log.warn("[{}:{}] Failed to add consumer handler for topic {}", request.getRemoteAddr(),
request.getRemotePort(), topic);
@@ -298,7 +297,7 @@ private ConsumerConfiguration getConsumerConfiguration() {
@Override
protected Boolean isAuthorized(String authRole, AuthenticationDataSource authenticationData) throws Exception {
- return service.getAuthorizationService().canConsume(TopicName.get(topic), authRole, authenticationData,
+ return service.getAuthorizationService().canConsume(topic, authRole, authenticationData,
this.subscription);
}
@@ -306,11 +305,18 @@ private static String extractSubscription(HttpServletRequest request) {
String uri = request.getRequestURI();
List<String> parts = Splitter.on("/").splitToList(uri);
- // Format must be like :
+ // v1 Format must be like :
// /ws/consumer/persistent/my-property/my-cluster/my-ns/my-topic/my-subscription
+
+ // v2 Format must be like :
+ // /ws/v2/consumer/persistent/my-property/my-ns/my-topic/my-subscription
checkArgument(parts.size() == 9, "Invalid topic name format");
checkArgument(parts.get(1).equals("ws"));
- checkArgument(parts.get(3).equals("persistent")|| parts.get(3).equals("non-persistent"));
+
+ final boolean isV2Format = parts.get(2).equals("v2");
+ final int domainIndex = isV2Format ? 4 : 3;
+ checkArgument(parts.get(domainIndex).equals("persistent") ||
+ parts.get(domainIndex).equals("non-persistent"));
checkArgument(parts.get(8).length() > 0, "Empty subscription name");
return parts.get(8);
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 151bdbf11..25fed37f1 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
@@ -44,7 +44,6 @@
import org.apache.pulsar.client.api.PulsarClientException.ProducerBlockedQuotaExceededError;
import org.apache.pulsar.client.api.PulsarClientException.ProducerBlockedQuotaExceededException;
import org.apache.pulsar.client.api.PulsarClientException.ProducerBusyException;
-import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.util.ObjectMapperFactory;
import org.apache.pulsar.websocket.data.ProducerAck;
import org.apache.pulsar.websocket.data.ProducerMessage;
@@ -93,7 +92,7 @@ public ProducerHandler(WebSocketService service, HttpServletRequest request, Ser
try {
ProducerConfiguration conf = getProducerConfiguration();
- this.producer = service.getPulsarClient().createProducer(topic, conf);
+ this.producer = service.getPulsarClient().createProducer(topic.toString(), conf);
if (!this.service.addProducer(this)) {
log.warn("[{}:{}] Failed to add producer handler for topic {}", request.getRemoteAddr(),
request.getRemotePort(), topic);
@@ -226,7 +225,7 @@ public long getMsgPublishedCounter() {
@Override
protected Boolean isAuthorized(String authRole, AuthenticationDataSource authenticationData) throws Exception {
- return service.getAuthorizationService().canProduce(TopicName.get(topic), authRole, authenticationData);
+ return service.getAuthorizationService().canProduce(topic, authRole, authenticationData);
}
private void sendAckResponse(ProducerAck response) {
diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ReaderHandler.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ReaderHandler.java
index 1eddd214f..2efb58545 100644
--- a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ReaderHandler.java
+++ b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ReaderHandler.java
@@ -32,11 +32,10 @@
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Reader;
-import org.apache.pulsar.client.api.ReaderConfiguration;
+import org.apache.pulsar.client.api.ReaderBuilder;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.impl.MessageIdImpl;
import org.apache.pulsar.client.impl.ReaderImpl;
-import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.util.DateFormatter;
import org.apache.pulsar.common.util.ObjectMapperFactory;
import org.apache.pulsar.websocket.data.ConsumerMessage;
@@ -55,8 +54,10 @@
*
*/
public class ReaderHandler extends AbstractWebSocketHandler {
- private String subscription;
- private final ReaderConfiguration conf;
+
+ private static final int DEFAULT_RECEIVER_QUEUE_SIZE = 1000;
+
+ private String subscription = "";
private Reader<byte[]> reader;
private final int maxPendingMessages;
@@ -70,9 +71,10 @@
public ReaderHandler(WebSocketService service, HttpServletRequest request, ServletUpgradeResponse response) {
super(service, request, response);
- this.subscription = "";
- this.conf = getReaderConfiguration();
- this.maxPendingMessages = (conf.getReceiverQueueSize() == 0) ? 1 : conf.getReceiverQueueSize();
+
+ final int receiverQueueSize = getReceiverQueueSize();
+
+ this.maxPendingMessages = (receiverQueueSize == 0) ? 1 : receiverQueueSize;
this.numMsgsDelivered = new LongAdder();
this.numBytesDelivered = new LongAdder();
@@ -81,7 +83,16 @@ public ReaderHandler(WebSocketService service, HttpServletRequest request, Servl
}
try {
- this.reader = service.getPulsarClient().createReader(topic, getMessageId(), conf);
+ ReaderBuilder<byte[]> builder = service.getPulsarClient().newReader()
+ .topic(topic.toString())
+ .startMessageId(getMessageId())
+ .receiverQueueSize(receiverQueueSize);
+ if (queryParams.containsKey("readerName")) {
+ builder.readerName(queryParams.get("readerName"));
+ }
+
+ this.reader = builder.create();
+
this.subscription = ((ReaderImpl)this.reader).getConsumer().getSubscription();
if (!this.service.addReader(this)) {
log.warn("[{}:{}] Failed to add reader handler for topic {}", request.getRemoteAddr(),
@@ -199,7 +210,7 @@ public void close() throws IOException {
}
public Consumer getConsumer() {
- return ((ReaderImpl)reader).getConsumer();
+ return reader != null ? ((ReaderImpl)reader).getConsumer() : null;
}
public String getSubscription() {
@@ -228,25 +239,20 @@ protected void updateDeliverMsgStat(long msgSize) {
numBytesDelivered.add(msgSize);
}
- private ReaderConfiguration getReaderConfiguration() {
- ReaderConfiguration conf = new ReaderConfiguration();
-
- if (queryParams.containsKey("readerName")) {
- conf.setReaderName(queryParams.get("readerName"));
- }
-
- if (queryParams.containsKey("receiverQueueSize")) {
- conf.setReceiverQueueSize(Math.min(Integer.parseInt(queryParams.get("receiverQueueSize")), 1000));
- }
- return conf;
- }
-
@Override
protected Boolean isAuthorized(String authRole, AuthenticationDataSource authenticationData) throws Exception {
- return service.getAuthorizationService().canConsume(TopicName.get(topic), authRole, authenticationData,
+ return service.getAuthorizationService().canConsume(topic, authRole, authenticationData,
this.subscription);
}
+ private int getReceiverQueueSize() {
+ int size = DEFAULT_RECEIVER_QUEUE_SIZE;
+ if (queryParams.containsKey("receiverQueueSize")) {
+ size = Math.min(Integer.parseInt(queryParams.get("receiverQueueSize")), DEFAULT_RECEIVER_QUEUE_SIZE);
+ }
+ return size;
+ }
+
private MessageId getMessageId() throws IOException {
MessageId messageId = MessageId.latest;
if (isNotBlank(queryParams.get("messageId"))) {
diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketConsumerServlet.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketConsumerServlet.java
index 73c70dd82..68212363b 100644
--- a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketConsumerServlet.java
+++ b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketConsumerServlet.java
@@ -25,6 +25,7 @@
private static final long serialVersionUID = 1L;
public static final String SERVLET_PATH = "/ws/consumer";
+ public static final String SERVLET_PATH_V2 = "/ws/v2/consumer";
WebSocketService service;
diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketProducerServlet.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketProducerServlet.java
index 2dfc3770e..c533166d9 100644
--- a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketProducerServlet.java
+++ b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketProducerServlet.java
@@ -25,6 +25,7 @@
private static final long serialVersionUID = 1L;
public static final String SERVLET_PATH = "/ws/producer";
+ public static final String SERVLET_PATH_V2 = "/ws/v2/producer";
private final WebSocketService service;
diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketReaderServlet.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketReaderServlet.java
index aab0c62bc..6d80a038d 100644
--- a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketReaderServlet.java
+++ b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketReaderServlet.java
@@ -25,6 +25,7 @@
private static final long serialVersionUID = 1L;
public static final String SERVLET_PATH = "/ws/reader";
+ public static final String SERVLET_PATH_V2 = "/ws/v2/reader";
WebSocketService service;
----------------------------------------------------------------
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