You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by pe...@apache.org on 2021/07/07 13:18:48 UTC
[pulsar] branch branch-2.7 updated (dd6e247 -> 6ba062c)
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a change to branch branch-2.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git.
from dd6e247 Remove unused listener to reduce creating executor pool. (#11215)
new 208ca5c Fix init WebSocketService with ClusterData (#11234)
new 341ac71 1. Fix websocket tls bug. (#11243)
new 6ba062c Fix cherry-pick issue.
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../pulsar/proxy/server/ProxyServiceStarter.java | 123 +++++++++------
.../proxy/server/ProxyServiceStarterTest.java | 152 +++++++++++++++++++
.../proxy/server/ProxyServiceTlsStarterTest.java | 165 +++++++++++++++++++++
.../src/test/resources}/proxy.conf | 10 +-
4 files changed, 403 insertions(+), 47 deletions(-)
create mode 100644 pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyServiceStarterTest.java
create mode 100644 pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyServiceTlsStarterTest.java
copy {conf => pulsar-proxy/src/test/resources}/proxy.conf (96%)
[pulsar] 02/03: 1. Fix websocket tls bug. (#11243)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 341ac715aa7a9184988064c6089d89a35df5c07f
Author: Tboy <te...@apache.org>
AuthorDate: Wed Jul 7 20:32:53 2021 +0800
1. Fix websocket tls bug. (#11243)
2. Add ProxyServiceTlsStarterTest.
3. Refactor ProxyServiceStarter to be more easy to test.
(cherry picked from commit 887d74f902211e39a38257b9f3099d2acd777418)
---
.../pulsar/proxy/server/ProxyServiceStarter.java | 33 ++++++++++++-----
.../proxy/server/ProxyServiceStarterTest.java | 8 +++--
...erTest.java => ProxyServiceTlsStarterTest.java} | 41 ++++++++++++----------
3 files changed, 53 insertions(+), 29 deletions(-)
diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyServiceStarter.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyServiceStarter.java
index c34f8cc..300091d 100644
--- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyServiceStarter.java
+++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyServiceStarter.java
@@ -28,6 +28,7 @@ import static org.slf4j.bridge.SLF4JBridgeHandler.removeHandlersForRootLogger;
import com.google.common.annotations.VisibleForTesting;
import org.apache.pulsar.broker.PulsarServerException;
+import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.authentication.AuthenticationService;
import org.apache.pulsar.common.configuration.PulsarConfigurationLoader;
import org.apache.pulsar.broker.web.plugin.servlet.AdditionalServletWithClassLoader;
@@ -86,6 +87,10 @@ public class ProxyServiceStarter {
private ProxyConfiguration config;
+ private ProxyService proxyService;
+
+ private WebServer server;
+
public ProxyServiceStarter(String[] args) throws Exception {
try {
@@ -155,17 +160,12 @@ public class ProxyServiceStarter {
AuthenticationService authenticationService = new AuthenticationService(
PulsarConfigurationLoader.convertFrom(config));
// create proxy service
- ProxyService proxyService = new ProxyService(config, authenticationService);
+ proxyService = new ProxyService(config, authenticationService);
// create a web-service
- final WebServer server = new WebServer(config, authenticationService);
+ server = new WebServer(config, authenticationService);
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
- try {
- proxyService.close();
- server.stop();
- } catch (Exception e) {
- log.warn("server couldn't stop gracefully {}", e.getMessage(), e);
- }
+ close();
}));
proxyService.start();
@@ -194,6 +194,19 @@ public class ProxyServiceStarter {
server.start();
}
+ public void close() {
+ try {
+ if(proxyService != null) {
+ proxyService.close();
+ }
+ if(server != null) {
+ server.stop();
+ }
+ } catch (Exception e) {
+ log.warn("server couldn't stop gracefully {}", e.getMessage(), e);
+ }
+ }
+
public static void addWebServerHandlers(WebServer server,
ProxyConfiguration config,
ProxyService service,
@@ -232,7 +245,9 @@ public class ProxyServiceStarter {
if (config.isWebSocketServiceEnabled()) {
// add WebSocket servlet
// Use local broker address to avoid different IP address when using a VIP for service discovery
- WebSocketService webSocketService = new WebSocketService(createClusterData(config), PulsarConfigurationLoader.convertFrom(config));
+ ServiceConfiguration serviceConfiguration = PulsarConfigurationLoader.convertFrom(config);
+ serviceConfiguration.setBrokerClientTlsEnabled(config.isTlsEnabledWithBroker());
+ WebSocketService webSocketService = new WebSocketService(createClusterData(config), serviceConfiguration);
webSocketService.start();
final WebSocketServlet producerWebSocketServlet = new WebSocketProducerServlet(webSocketService);
server.addServlet(WebSocketProducerServlet.SERVLET_PATH,
diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyServiceStarterTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyServiceStarterTest.java
index 63d7e39..3377ec2 100644
--- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyServiceStarterTest.java
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyServiceStarterTest.java
@@ -48,12 +48,15 @@ import static org.testng.Assert.assertTrue;
public class ProxyServiceStarterTest extends MockedPulsarServiceBaseTest {
+ static final String[] ARGS = new String[]{"-c", "./src/test/resources/proxy.conf"};
+
+ private ProxyServiceStarter serviceStarter;
+
@Override
@BeforeClass
protected void setup() throws Exception {
internalSetup();
- String[] args = new String[]{"-c", "./src/test/resources/proxy.conf"};
- ProxyServiceStarter serviceStarter = new ProxyServiceStarter(args);
+ serviceStarter = new ProxyServiceStarter(ARGS);
serviceStarter.getConfig().setBrokerServiceURL(pulsar.getBrokerServiceUrl());
serviceStarter.getConfig().setBrokerWebServiceURL(pulsar.getWebServiceAddress());
serviceStarter.getConfig().setServicePort(Optional.of(11000));
@@ -65,6 +68,7 @@ public class ProxyServiceStarterTest extends MockedPulsarServiceBaseTest {
@AfterClass(alwaysRun = true)
protected void cleanup() throws Exception {
internalCleanup();
+ serviceStarter.close();
}
@Test
diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyServiceStarterTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyServiceTlsStarterTest.java
similarity index 79%
copy from pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyServiceStarterTest.java
copy to pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyServiceTlsStarterTest.java
index 63d7e39..7e6c0f5 100644
--- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyServiceStarterTest.java
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyServiceTlsStarterTest.java
@@ -20,7 +20,6 @@ package org.apache.pulsar.proxy.server;
import lombok.Cleanup;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
-
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.common.util.ObjectMapperFactory;
@@ -43,46 +42,52 @@ import java.util.Optional;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Future;
+import static org.apache.pulsar.proxy.server.ProxyServiceStarterTest.ARGS;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertTrue;
-public class ProxyServiceStarterTest extends MockedPulsarServiceBaseTest {
+public class ProxyServiceTlsStarterTest extends MockedPulsarServiceBaseTest {
+
+ private final String TLS_TRUST_CERT_FILE_PATH = "./src/test/resources/authentication/tls/cacert.pem";
+ private final String TLS_PROXY_CERT_FILE_PATH = "./src/test/resources/authentication/tls/server-cert.pem";
+ private final String TLS_PROXY_KEY_FILE_PATH = "./src/test/resources/authentication/tls/server-key.pem";
+ private ProxyServiceStarter serviceStarter;
@Override
@BeforeClass
protected void setup() throws Exception {
internalSetup();
- String[] args = new String[]{"-c", "./src/test/resources/proxy.conf"};
- ProxyServiceStarter serviceStarter = new ProxyServiceStarter(args);
+ serviceStarter = new ProxyServiceStarter(ARGS);
serviceStarter.getConfig().setBrokerServiceURL(pulsar.getBrokerServiceUrl());
+ serviceStarter.getConfig().setBrokerServiceURLTLS(pulsar.getBrokerServiceUrlTls());
serviceStarter.getConfig().setBrokerWebServiceURL(pulsar.getWebServiceAddress());
- serviceStarter.getConfig().setServicePort(Optional.of(11000));
+ serviceStarter.getConfig().setBrokerClientTrustCertsFilePath(TLS_TRUST_CERT_FILE_PATH);
+ serviceStarter.getConfig().setServicePortTls(Optional.of(11043));
+ serviceStarter.getConfig().setTlsEnabledWithBroker(true);
serviceStarter.getConfig().setWebSocketServiceEnabled(true);
+ serviceStarter.getConfig().setTlsCertificateFilePath(TLS_PROXY_CERT_FILE_PATH);
+ serviceStarter.getConfig().setTlsKeyFilePath(TLS_PROXY_KEY_FILE_PATH);
serviceStarter.start();
}
+ protected void doInitConf() throws Exception {
+ super.doInitConf();
+ this.conf.setTlsCertificateFilePath(TLS_PROXY_CERT_FILE_PATH);
+ this.conf.setTlsKeyFilePath(TLS_PROXY_KEY_FILE_PATH);
+ }
+
@Override
@AfterClass(alwaysRun = true)
protected void cleanup() throws Exception {
internalCleanup();
- }
-
- @Test
- public void testEnableWebSocketServer() throws Exception {
- HttpClient httpClient = new HttpClient();
- WebSocketClient webSocketClient = new WebSocketClient(httpClient);
- webSocketClient.start();
- MyWebSocket myWebSocket = new MyWebSocket();
- String webSocketUri = "ws://localhost:8080/ws/pingpong";
- Future<Session> sessionFuture = webSocketClient.connect(myWebSocket, URI.create(webSocketUri));
- sessionFuture.get().getRemote().sendPing(ByteBuffer.wrap("ping".getBytes()));
- assertTrue(myWebSocket.getResponse().contains("ping"));
+ serviceStarter.close();
}
@Test
public void testProducer() throws Exception {
@Cleanup
- PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://localhost:11000")
+ PulsarClient client = PulsarClient.builder().serviceUrl("pulsar+ssl://localhost:11043")
+ .allowTlsInsecureConnection(false).tlsTrustCertsFilePath(TLS_TRUST_CERT_FILE_PATH)
.build();
@Cleanup
[pulsar] 03/03: Fix cherry-pick issue.
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 6ba062c11317e07d7dca7863f56e9ac8d116047b
Author: penghui <pe...@apache.org>
AuthorDate: Wed Jul 7 21:17:27 2021 +0800
Fix cherry-pick issue.
---
.../org/apache/pulsar/proxy/server/ProxyServiceStarter.java | 13 +++----------
.../apache/pulsar/proxy/server/ProxyServiceStarterTest.java | 12 ------------
2 files changed, 3 insertions(+), 22 deletions(-)
diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyServiceStarter.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyServiceStarter.java
index 300091d..25e0f59 100644
--- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyServiceStarter.java
+++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyServiceStarter.java
@@ -271,17 +271,10 @@ public class ProxyServiceStarter {
private static ClusterData createClusterData(ProxyConfiguration config) {
if (isNotBlank(config.getBrokerServiceURL()) || isNotBlank(config.getBrokerServiceURLTLS())) {
- return ClusterData.builder()
- .serviceUrl(config.getBrokerWebServiceURL())
- .serviceUrlTls(config.getBrokerWebServiceURLTLS())
- .brokerServiceUrl(config.getBrokerServiceURL())
- .brokerServiceUrlTls(config.getBrokerServiceURLTLS())
- .build();
+ return new ClusterData(config.getBrokerWebServiceURL(), config.getBrokerWebServiceURLTLS(),
+ config.getBrokerServiceURL(), config.getBrokerServiceURLTLS());
} else if (isNotBlank(config.getBrokerWebServiceURL()) || isNotBlank(config.getBrokerWebServiceURLTLS())) {
- return ClusterData.builder()
- .serviceUrl(config.getBrokerWebServiceURL())
- .serviceUrlTls(config.getBrokerWebServiceURLTLS())
- .build();
+ return new ClusterData(config.getBrokerWebServiceURL(), config.getBrokerWebServiceURLTLS());
} else {
return null;
}
diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyServiceStarterTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyServiceStarterTest.java
index 3377ec2..228927f 100644
--- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyServiceStarterTest.java
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyServiceStarterTest.java
@@ -72,18 +72,6 @@ public class ProxyServiceStarterTest extends MockedPulsarServiceBaseTest {
}
@Test
- public void testEnableWebSocketServer() throws Exception {
- HttpClient httpClient = new HttpClient();
- WebSocketClient webSocketClient = new WebSocketClient(httpClient);
- webSocketClient.start();
- MyWebSocket myWebSocket = new MyWebSocket();
- String webSocketUri = "ws://localhost:8080/ws/pingpong";
- Future<Session> sessionFuture = webSocketClient.connect(myWebSocket, URI.create(webSocketUri));
- sessionFuture.get().getRemote().sendPing(ByteBuffer.wrap("ping".getBytes()));
- assertTrue(myWebSocket.getResponse().contains("ping"));
- }
-
- @Test
public void testProducer() throws Exception {
@Cleanup
PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://localhost:11000")
[pulsar] 01/03: Fix init WebSocketService with ClusterData (#11234)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.7
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 208ca5c6fbe79b73c5f4b5d744a8d40694a6175f
Author: Tboy <te...@apache.org>
AuthorDate: Wed Jul 7 07:45:03 2021 +0800
Fix init WebSocketService with ClusterData (#11234)
1. Fix init WebSocketService with ClusterData.
2. Refactor ProxyServiceStarter to be easier for test.
3. Add test case for ProxyServiceStarter.
(cherry picked from commit 93e145bea96c88a24d33cf65e657e86ac4e4c491)
---
.../pulsar/proxy/server/ProxyServiceStarter.java | 117 ++++++----
.../proxy/server/ProxyServiceStarterTest.java | 160 ++++++++++++++
pulsar-proxy/src/test/resources/proxy.conf | 236 +++++++++++++++++++++
3 files changed, 470 insertions(+), 43 deletions(-)
diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyServiceStarter.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyServiceStarter.java
index 3e79fe2..c34f8cc 100644
--- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyServiceStarter.java
+++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyServiceStarter.java
@@ -21,14 +21,17 @@ package org.apache.pulsar.proxy.server;
import static com.google.common.base.Preconditions.checkArgument;
import static org.apache.commons.lang3.StringUtils.isBlank;
import static org.apache.commons.lang3.StringUtils.isEmpty;
+import static org.apache.commons.lang3.StringUtils.isNotBlank;
import static org.apache.pulsar.common.stats.JvmMetrics.getJvmDirectMemoryUsed;
import static org.slf4j.bridge.SLF4JBridgeHandler.install;
import static org.slf4j.bridge.SLF4JBridgeHandler.removeHandlersForRootLogger;
+import com.google.common.annotations.VisibleForTesting;
import org.apache.pulsar.broker.PulsarServerException;
import org.apache.pulsar.broker.authentication.AuthenticationService;
import org.apache.pulsar.common.configuration.PulsarConfigurationLoader;
import org.apache.pulsar.broker.web.plugin.servlet.AdditionalServletWithClassLoader;
+import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.websocket.WebSocketConsumerServlet;
import org.apache.pulsar.websocket.WebSocketProducerServlet;
import org.apache.pulsar.websocket.WebSocketReaderServlet;
@@ -81,6 +84,8 @@ public class ProxyServiceStarter {
@Parameter(names = { "-h", "--help" }, description = "Show this help message")
private boolean help = false;
+ private ProxyConfiguration config;
+
public ProxyServiceStarter(String[] args) throws Exception {
try {
@@ -107,7 +112,7 @@ public class ProxyServiceStarter {
}
// load config file
- final ProxyConfiguration config = PulsarConfigurationLoader.create(configFile, ProxyConfiguration.class);
+ config = PulsarConfigurationLoader.create(configFile, ProxyConfiguration.class);
if (!isBlank(zookeeperServers)) {
// Use zookeeperServers from command line
@@ -135,55 +140,58 @@ public class ProxyServiceStarter {
checkArgument(!isEmpty(config.getZookeeperServers()), "zookeeperServers must be provided");
}
- AuthenticationService authenticationService = new AuthenticationService(
- PulsarConfigurationLoader.convertFrom(config));
- // create proxy service
- ProxyService proxyService = new ProxyService(config, authenticationService);
- // create a web-service
- final WebServer server = new WebServer(config, authenticationService);
-
- Runtime.getRuntime().addShutdownHook(new Thread(() -> {
- try {
- proxyService.close();
- server.stop();
- } catch (Exception e) {
- log.warn("server couldn't stop gracefully {}", e.getMessage(), e);
- }
- }));
+ } catch (Exception e) {
+ log.error("Failed to start pulsar proxy service. error msg " + e.getMessage(), e);
+ throw new PulsarServerException(e);
+ }
+ }
- proxyService.start();
+ public static void main(String[] args) throws Exception {
+ ProxyServiceStarter serviceStarter = new ProxyServiceStarter(args);
+ serviceStarter.start();
+ }
- // Setup metrics
- DefaultExports.initialize();
+ public void start() throws Exception {
+ AuthenticationService authenticationService = new AuthenticationService(
+ PulsarConfigurationLoader.convertFrom(config));
+ // create proxy service
+ ProxyService proxyService = new ProxyService(config, authenticationService);
+ // create a web-service
+ final WebServer server = new WebServer(config, authenticationService);
- // Report direct memory from Netty counters
- Gauge.build("jvm_memory_direct_bytes_used", "-").create().setChild(new Child() {
- @Override
- public double get() {
- return getJvmDirectMemoryUsed();
- }
- }).register(CollectorRegistry.defaultRegistry);
+ Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+ try {
+ proxyService.close();
+ server.stop();
+ } catch (Exception e) {
+ log.warn("server couldn't stop gracefully {}", e.getMessage(), e);
+ }
+ }));
- Gauge.build("jvm_memory_direct_bytes_max", "-").create().setChild(new Child() {
- @Override
- public double get() {
- return PlatformDependent.maxDirectMemory();
- }
- }).register(CollectorRegistry.defaultRegistry);
+ proxyService.start();
- addWebServerHandlers(server, config, proxyService, proxyService.getDiscoveryProvider());
+ // Setup metrics
+ DefaultExports.initialize();
- // start web-service
- server.start();
+ // Report direct memory from Netty counters
+ Gauge.build("jvm_memory_direct_bytes_used", "-").create().setChild(new Child() {
+ @Override
+ public double get() {
+ return getJvmDirectMemoryUsed();
+ }
+ }).register(CollectorRegistry.defaultRegistry);
- } catch (Exception e) {
- log.error("Failed to start pulsar proxy service. error msg " + e.getMessage(), e);
- throw new PulsarServerException(e);
- }
- }
+ Gauge.build("jvm_memory_direct_bytes_max", "-").create().setChild(new Child() {
+ @Override
+ public double get() {
+ return PlatformDependent.maxDirectMemory();
+ }
+ }).register(CollectorRegistry.defaultRegistry);
- public static void main(String[] args) throws Exception {
- new ProxyServiceStarter(args);
+ addWebServerHandlers(server, config, proxyService, proxyService.getDiscoveryProvider());
+
+ // start web-service
+ server.start();
}
public static void addWebServerHandlers(WebServer server,
@@ -224,7 +232,7 @@ public class ProxyServiceStarter {
if (config.isWebSocketServiceEnabled()) {
// add WebSocket servlet
// Use local broker address to avoid different IP address when using a VIP for service discovery
- WebSocketService webSocketService = new WebSocketService(null, PulsarConfigurationLoader.convertFrom(config));
+ WebSocketService webSocketService = new WebSocketService(createClusterData(config), PulsarConfigurationLoader.convertFrom(config));
webSocketService.start();
final WebSocketServlet producerWebSocketServlet = new WebSocketProducerServlet(webSocketService);
server.addServlet(WebSocketProducerServlet.SERVLET_PATH,
@@ -246,6 +254,29 @@ public class ProxyServiceStarter {
}
}
+ private static ClusterData createClusterData(ProxyConfiguration config) {
+ if (isNotBlank(config.getBrokerServiceURL()) || isNotBlank(config.getBrokerServiceURLTLS())) {
+ return ClusterData.builder()
+ .serviceUrl(config.getBrokerWebServiceURL())
+ .serviceUrlTls(config.getBrokerWebServiceURLTLS())
+ .brokerServiceUrl(config.getBrokerServiceURL())
+ .brokerServiceUrlTls(config.getBrokerServiceURLTLS())
+ .build();
+ } else if (isNotBlank(config.getBrokerWebServiceURL()) || isNotBlank(config.getBrokerWebServiceURLTLS())) {
+ return ClusterData.builder()
+ .serviceUrl(config.getBrokerWebServiceURL())
+ .serviceUrlTls(config.getBrokerWebServiceURLTLS())
+ .build();
+ } else {
+ return null;
+ }
+ }
+
+ @VisibleForTesting
+ public ProxyConfiguration getConfig() {
+ return config;
+ }
+
private static final Logger log = LoggerFactory.getLogger(ProxyServiceStarter.class);
}
diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyServiceStarterTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyServiceStarterTest.java
new file mode 100644
index 0000000..63d7e39
--- /dev/null
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyServiceStarterTest.java
@@ -0,0 +1,160 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.proxy.server;
+
+import lombok.Cleanup;
+import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.common.util.ObjectMapperFactory;
+import org.apache.pulsar.websocket.data.ProducerMessage;
+import org.eclipse.jetty.client.HttpClient;
+import org.eclipse.jetty.util.BufferUtil;
+import org.eclipse.jetty.websocket.api.Session;
+import org.eclipse.jetty.websocket.api.WebSocketAdapter;
+import org.eclipse.jetty.websocket.api.WebSocketPingPongListener;
+import org.eclipse.jetty.websocket.api.annotations.WebSocket;
+import org.eclipse.jetty.websocket.client.WebSocketClient;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import java.net.URI;
+import java.nio.ByteBuffer;
+import java.util.Base64;
+import java.util.Optional;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.Future;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+
+public class ProxyServiceStarterTest extends MockedPulsarServiceBaseTest {
+
+ @Override
+ @BeforeClass
+ protected void setup() throws Exception {
+ internalSetup();
+ String[] args = new String[]{"-c", "./src/test/resources/proxy.conf"};
+ ProxyServiceStarter serviceStarter = new ProxyServiceStarter(args);
+ serviceStarter.getConfig().setBrokerServiceURL(pulsar.getBrokerServiceUrl());
+ serviceStarter.getConfig().setBrokerWebServiceURL(pulsar.getWebServiceAddress());
+ serviceStarter.getConfig().setServicePort(Optional.of(11000));
+ serviceStarter.getConfig().setWebSocketServiceEnabled(true);
+ serviceStarter.start();
+ }
+
+ @Override
+ @AfterClass(alwaysRun = true)
+ protected void cleanup() throws Exception {
+ internalCleanup();
+ }
+
+ @Test
+ public void testEnableWebSocketServer() throws Exception {
+ HttpClient httpClient = new HttpClient();
+ WebSocketClient webSocketClient = new WebSocketClient(httpClient);
+ webSocketClient.start();
+ MyWebSocket myWebSocket = new MyWebSocket();
+ String webSocketUri = "ws://localhost:8080/ws/pingpong";
+ Future<Session> sessionFuture = webSocketClient.connect(myWebSocket, URI.create(webSocketUri));
+ sessionFuture.get().getRemote().sendPing(ByteBuffer.wrap("ping".getBytes()));
+ assertTrue(myWebSocket.getResponse().contains("ping"));
+ }
+
+ @Test
+ public void testProducer() throws Exception {
+ @Cleanup
+ PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://localhost:11000")
+ .build();
+
+ @Cleanup
+ Producer<byte[]> producer = client.newProducer()
+ .topic("persistent://sample/test/local/websocket-topic")
+ .create();
+
+ for (int i = 0; i < 10; i++) {
+ producer.send("test".getBytes());
+ }
+ }
+
+ @Test
+ public void testProduceAndConsumeMessageWithWebsocket() throws Exception {
+ HttpClient producerClient = new HttpClient();
+ WebSocketClient producerWebSocketClient = new WebSocketClient(producerClient);
+ producerWebSocketClient.start();
+ MyWebSocket producerSocket = new MyWebSocket();
+ String produceUri = "ws://localhost:8080/ws/producer/persistent/sample/test/local/websocket-topic";
+ Future<Session> producerSession = producerWebSocketClient.connect(producerSocket, URI.create(produceUri));
+
+ ProducerMessage produceRequest = new ProducerMessage();
+ produceRequest.setContext("context");
+ produceRequest.setPayload(Base64.getEncoder().encodeToString("my payload".getBytes()));
+
+ HttpClient consumerClient = new HttpClient();
+ WebSocketClient consumerWebSocketClient = new WebSocketClient(consumerClient);
+ consumerWebSocketClient.start();
+ MyWebSocket consumerSocket = new MyWebSocket();
+ String consumeUri = "ws://localhost:8080/ws/consumer/persistent/sample/test/local/websocket-topic/my-sub";
+ Future<Session> consumerSession = consumerWebSocketClient.connect(consumerSocket, URI.create(consumeUri));
+ consumerSession.get().getRemote().sendPing(ByteBuffer.wrap("ping".getBytes()));
+ producerSession.get().getRemote().sendString(ObjectMapperFactory.getThreadLocal().writeValueAsString(produceRequest));
+ assertTrue(consumerSocket.getResponse().contains("ping"));
+ ProducerMessage message = ObjectMapperFactory.getThreadLocal().readValue(consumerSocket.getResponse(), ProducerMessage.class);
+ assertEquals(new String(Base64.getDecoder().decode(message.getPayload())), "my payload");
+ }
+
+ @WebSocket
+ public static class MyWebSocket extends WebSocketAdapter implements WebSocketPingPongListener {
+
+ ArrayBlockingQueue<String> incomingMessages = new ArrayBlockingQueue<>(10);
+
+ @Override
+ public void onWebSocketText(String message) {
+ incomingMessages.add(message);
+ }
+
+ @Override
+ public void onWebSocketClose(int i, String s) {
+ }
+
+ @Override
+ public void onWebSocketConnect(Session session) {
+ }
+
+ @Override
+ public void onWebSocketError(Throwable throwable) {
+ }
+
+ @Override
+ public void onWebSocketPing(ByteBuffer payload) {
+ }
+
+ @Override
+ public void onWebSocketPong(ByteBuffer payload) {
+ incomingMessages.add(BufferUtil.toDetailString(payload));
+ }
+
+ public String getResponse() throws InterruptedException {
+ return incomingMessages.take();
+ }
+ }
+
+}
diff --git a/pulsar-proxy/src/test/resources/proxy.conf b/pulsar-proxy/src/test/resources/proxy.conf
new file mode 100644
index 0000000..b5ed33f
--- /dev/null
+++ b/pulsar-proxy/src/test/resources/proxy.conf
@@ -0,0 +1,236 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+### --- Broker Discovery --- ###
+
+# The ZooKeeper quorum connection string (as a comma-separated list)
+zookeeperServers=
+
+# Configuration store connection string (as a comma-separated list)
+configurationStoreServers=
+
+# if Service Discovery is Disabled this url should point to the discovery service provider.
+brokerServiceURL=pulsar://0.0.0.0:0
+brokerServiceURLTLS=
+
+# These settings are unnecessary if `zookeeperServers` is specified
+brokerWebServiceURL=http://0.0.0.0:0
+brokerWebServiceURLTLS=
+
+# If function workers are setup in a separate cluster, configure the following 2 settings
+# to point to the function workers cluster
+functionWorkerWebServiceURL=
+functionWorkerWebServiceURLTLS=
+
+# ZooKeeper session timeout (in milliseconds)
+zookeeperSessionTimeoutMs=30000
+
+# ZooKeeper cache expiry time in seconds
+zooKeeperCacheExpirySeconds=300
+
+### --- Server --- ###
+
+# Hostname or IP address the service binds on, default is 0.0.0.0.
+bindAddress=0.0.0.0
+
+# Hostname or IP address the service advertises to the outside world.
+# If not set, the value of `InetAddress.getLocalHost().getHostname()` is used.
+advertisedAddress=
+
+# Enable or disable the HAProxy protocol.
+haProxyProtocolEnabled=false
+
+# The port to use for server binary Protobuf requests
+servicePort=6650
+
+# The port to use to server binary Protobuf TLS requests
+servicePortTls=
+
+# Port that discovery service listen on
+webServicePort=8080
+
+# Port to use to server HTTPS request
+webServicePortTls=
+
+# Path for the file used to determine the rotation status for the proxy instance when responding
+# to service discovery health checks
+statusFilePath=
+
+# Proxy log level, default is 0.
+# 0: Do not log any tcp channel info
+# 1: Parse and log any tcp channel info and command info without message body
+# 2: Parse and log channel info, command info and message body
+proxyLogLevel=0
+
+### ---Authorization --- ###
+
+# Role names that are treated as "super-users," meaning that they will be able to perform all admin
+# operations and publish/consume to/from all topics (as a comma-separated list)
+superUserRoles=
+
+# Whether authorization is enforced by the Pulsar proxy
+authorizationEnabled=false
+
+# Authorization provider as a fully qualified class name
+authorizationProvider=org.apache.pulsar.broker.authorization.PulsarAuthorizationProvider
+
+# Whether client authorization credentials are forwared to the broker for re-authorization.
+# Authentication must be enabled via authenticationEnabled=true for this to take effect.
+forwardAuthorizationCredentials=false
+
+### --- Authentication --- ###
+
+# Whether authentication is enabled for the Pulsar proxy
+authenticationEnabled=false
+
+# Authentication provider name list (a comma-separated list of class names)
+authenticationProviders=
+
+# When this parameter is not empty, unauthenticated users perform as anonymousUserRole
+anonymousUserRole=
+
+### --- Client Authentication --- ###
+
+# The three brokerClient* authentication settings below are for the proxy itself and determine how it
+# authenticates with Pulsar brokers
+
+# The authentication plugin used by the Pulsar proxy to authenticate with Pulsar brokers
+brokerClientAuthenticationPlugin=
+
+# The authentication parameters used by the Pulsar proxy to authenticate with Pulsar brokers
+brokerClientAuthenticationParameters=
+
+# The path to trusted certificates used by the Pulsar proxy to authenticate with Pulsar brokers
+brokerClientTrustCertsFilePath=
+
+# Whether TLS is enabled when communicating with Pulsar brokers
+tlsEnabledWithBroker=false
+
+# Tls cert refresh duration in seconds (set 0 to check on every new connection)
+tlsCertRefreshCheckDurationSec=300
+
+##### --- Rate Limiting --- #####
+
+# Max concurrent inbound connections. The proxy will reject requests beyond that.
+maxConcurrentInboundConnections=10000
+
+# Max concurrent outbound connections. The proxy will error out requests beyond that.
+maxConcurrentLookupRequests=50000
+
+##### --- TLS --- #####
+
+# Deprecated - use servicePortTls and webServicePortTls instead
+tlsEnabledInProxy=false
+
+# Path for the TLS certificate file
+tlsCertificateFilePath=
+
+# Path for the TLS private key file
+tlsKeyFilePath=
+
+# Path for the trusted TLS certificate file.
+# This cert is used to verify that any certs presented by connecting clients
+# are signed by a certificate authority. If this verification
+# fails, then the certs are untrusted and the connections are dropped.
+tlsTrustCertsFilePath=
+
+# Accept untrusted TLS certificate from client.
+# If true, a client with a cert which cannot be verified with the
+# 'tlsTrustCertsFilePath' cert will allowed to connect to the server,
+# though the cert will not be used for client authentication.
+tlsAllowInsecureConnection=false
+
+# Whether the hostname is validated when the proxy creates a TLS connection with brokers
+tlsHostnameVerificationEnabled=false
+
+# Specify the tls protocols the broker will use to negotiate during TLS handshake
+# (a comma-separated list of protocol names).
+# Examples:- [TLSv1.3, TLSv1.2]
+tlsProtocols=
+
+# Specify the tls cipher the broker will use to negotiate during TLS Handshake
+# (a comma-separated list of ciphers).
+# Examples:- [TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256]
+tlsCiphers=
+
+# Whether client certificates are required for TLS. Connections are rejected if the client
+# certificate isn't trusted.
+tlsRequireTrustedClientCertOnConnect=false
+
+##### --- HTTP --- #####
+
+# Http directs to redirect to non-pulsar services.
+httpReverseProxyConfigs=
+
+# Http output buffer size. The amount of data that will be buffered for http requests
+# before it is flushed to the channel. A larger buffer size may result in higher http throughput
+# though it may take longer for the client to see data.
+# If using HTTP streaming via the reverse proxy, this should be set to the minimum value, 1,
+# so that clients see the data as soon as possible.
+httpOutputBufferSize=32768
+
+# Number of threads to use for HTTP requests processing. Default is
+# 2 * Runtime.getRuntime().availableProcessors()
+httpNumThreads=
+
+# Enable the enforcement of limits on the incoming HTTP requests
+httpRequestsLimitEnabled=false
+
+# Max HTTP requests per seconds allowed. The excess of requests will be rejected with HTTP code 429 (Too many requests)
+httpRequestsMaxPerSecond=100.0
+
+
+### --- Token Authentication Provider --- ###
+
+## Symmetric key
+# Configure the secret key to be used to validate auth tokens
+# The key can be specified like:
+# tokenSecretKey=data:;base64,xxxxxxxxx
+# tokenSecretKey=file:///my/secret.key ( Note: key file must be DER-encoded )
+tokenSecretKey=
+
+## Asymmetric public/private key pair
+# Configure the public key to be used to validate auth tokens
+# The key can be specified like:
+# tokenPublicKey=data:;base64,xxxxxxxxx
+# tokenPublicKey=file:///my/public.key ( Note: key file must be DER-encoded )
+tokenPublicKey=
+
+# The token "claim" that will be interpreted as the authentication "role" or "principal" by AuthenticationProviderToken (defaults to "sub" if blank)
+tokenAuthClaim=
+
+# The token audience "claim" name, e.g. "aud", that will be used to get the audience from token.
+# If not set, audience will not be verified.
+tokenAudienceClaim=
+
+# The token audience stands for this broker. The field `tokenAudienceClaim` of a valid token, need contains this.
+tokenAudience=
+
+### --- WebSocket config variables --- ###
+
+# Enable or disable the WebSocket servlet.
+webSocketServiceEnabled=false
+
+# Name of the cluster to which this broker belongs to
+clusterName=
+
+### --- Deprecated config variables --- ###
+
+# Deprecated. Use configurationStoreServers
+globalZookeeperServers=