Skip to content

Fix: broken websocket test and proxy-producer error handling #99

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merlimat merged 1 commit into from
Nov 1, 2016
Merged

Fix: broken websocket test and proxy-producer error handling #99

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions docs/WebSocket.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,13 +95,22 @@ http://{serviceUrl}:8080/ws/producer/persistent/{property}/{cluster}/{namespace}

##### Acknowledgement from server

###### Success response
```json
{
"result": "ok",
"messageId": "CAAQAw==",
"context": "1"
}
```
###### Failure response
```json
{
"result": "send-error:3",
"errorMsg": "Failed to de-serialize from JSON",
"context": "1"
}

@rdhabalia rdhabalia Nov 1, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

as doc doesn't have any failure response example: added failure response

@merlimat merlimat Nov 1, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok, but keep it in a separate code block, to make it easier to see that they are 2 different json responses.

@rdhabalia rdhabalia Nov 1, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

sure. done.

```

| Key | Type | Requirement | Explanation |
|:------------|:------:|:-----------:|:--------------------------------------------:|
Expand Down Expand Up @@ -172,6 +181,8 @@ following error codes:
| 4 | Failed to serialize to JSON |
| 5 | Failed to authenticate client |
| 6 | Client is not authorized |
| 7 | Invalid payload encoding |
| 8 | Unknown error |

Application is responsible to re-establish a new WebSocket session after
a backoff period.
Expand Down
2 changes: 1 addition & 1 deletion pulsar-broker/src/main/java/com/yahoo/pulsar/broker/admin/AdminResource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -237,7 +237,7 @@ protected Policies getNamespacePolicies(String property, String cluster, String
}
}

public ObjectMapper jsonMapper() {
public static ObjectMapper jsonMapper() {
return ObjectMapperFactory.getThreadLocal();
}

Expand Down
36 changes: 25 additions & 11 deletions pulsar-broker/src/test/java/com/yahoo/pulsar/websocket/proxy/ProxyAuthenticationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,50 +20,60 @@
import static org.mockito.Mockito.spy;

import java.net.URI;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import org.apache.bookkeeper.test.PortManager;
import org.eclipse.jetty.websocket.api.Session;
import org.eclipse.jetty.websocket.client.ClientUpgradeRequest;
import org.eclipse.jetty.websocket.client.WebSocketClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;

import com.google.common.collect.Sets;
import com.yahoo.pulsar.broker.ServiceConfiguration;
import com.yahoo.pulsar.broker.auth.MockedPulsarServiceBaseTest;
import com.yahoo.pulsar.client.api.ProducerConsumerBase;
import com.yahoo.pulsar.websocket.WebSocketService;
import com.yahoo.pulsar.websocket.service.ProxyServer;
import com.yahoo.pulsar.websocket.service.WebSocketServiceStarter;

public class ProxyAuthenticationTest extends MockedPulsarServiceBaseTest {
public class ProxyAuthenticationTest extends ProducerConsumerBase {
protected String methodName;
private static final int TEST_PORT = PortManager.nextFreePort();
private static final int TEST_PORT = 6080;
private static final String CONSUME_URI = "ws://localhost:" + TEST_PORT
+ "/consume/persistent/my-property/cluster1/my-ns/my-topic/my-sub";
+ "/ws/consumer/persistent/my-property/use/my-ns/my-topic/my-sub";
private static final String PRODUCE_URI = "ws://localhost:" + TEST_PORT
+ "/produce/persistent/my-property/cluster1/my-ns/my-topic/";
+ "/ws/producer/persistent/my-property/use/my-ns/my-topic/";
private ProxyServer proxyServer;
private WebSocketService service;

@BeforeClass
public void setup() throws Exception {
super.internalSetup();
super.producerBaseSetup();

ServiceConfiguration config = new ServiceConfiguration();
config.setWebServicePort(TEST_PORT);
config.setClusterName("use");
config.setAuthenticationEnabled(true);
config.setAuthenticationProviders(
Sets.newHashSet("com.yahoo.pulsar.websocket.proxy.MockAuthenticationProvider"));
service = spy(new WebSocketService(config));
doReturn(mockZooKeeperClientFactory).when(service).getZooKeeperClientFactory();
service.start();
proxyServer = new ProxyServer(config);
WebSocketServiceStarter.start(proxyServer, service);
log.info("Proxy Server Started");
}

@Override
@AfterClass
protected void cleanup() throws Exception {
super.internalCleanup();
service.close();
proxyServer.stop();
log.info("Finished Cleaning Up Test setup");

}
Expand All @@ -81,16 +91,20 @@ public void socketTest() throws InterruptedException {
try {
consumeClient.start();
ClientUpgradeRequest consumeRequest = new ClientUpgradeRequest();
consumeClient.connect(consumeSocket, consumeUri, consumeRequest);
log.info("Connecting to : %s%n", consumeUri);
Future<Session> consumerFuture = consumeClient.connect(consumeSocket, consumeUri, consumeRequest);
log.info("Connecting to : {}", consumeUri);

ClientUpgradeRequest produceRequest = new ClientUpgradeRequest();
produceClient.start();
produceClient.connect(produceSocket, produceUri, produceRequest);
Future<Session> producerFuture = produceClient.connect(produceSocket, produceUri, produceRequest);
// let it connect
Thread.sleep(1000);
Assert.assertTrue(consumerFuture.get().isOpen());
Assert.assertTrue(producerFuture.get().isOpen());

consumeSocket.awaitClose(1, TimeUnit.SECONDS);
produceSocket.awaitClose(1, TimeUnit.SECONDS);

Assert.assertTrue(produceSocket.getBuffer().size() > 0);
Assert.assertEquals(produceSocket.getBuffer(), consumeSocket.getBuffer());
} catch (Throwable t) {
log.error(t.getMessage());
Expand Down
41 changes: 26 additions & 15 deletions pulsar-broker/src/test/java/com/yahoo/pulsar/websocket/proxy/ProxyPublishConsumeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,49 +19,58 @@
import static org.mockito.Mockito.spy;

import java.net.URI;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import org.eclipse.jetty.websocket.api.Session;
import org.eclipse.jetty.websocket.client.ClientUpgradeRequest;
import org.eclipse.jetty.websocket.client.WebSocketClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;

import com.yahoo.pulsar.broker.ServiceConfiguration;
import com.yahoo.pulsar.broker.auth.MockedPulsarServiceBaseTest;
import com.yahoo.pulsar.client.api.ProducerConsumerBase;
import com.yahoo.pulsar.websocket.WebSocketService;
import com.yahoo.pulsar.websocket.service.ProxyServer;
import com.yahoo.pulsar.websocket.service.WebSocketServiceStarter;

public class ProxyPublishConsumeTest extends MockedPulsarServiceBaseTest {
public class ProxyPublishConsumeTest extends ProducerConsumerBase {
protected String methodName;
private static final String CONSUME_URI = "ws://localhost:6080/ws/consumer/persistent/my-property/cluster1/my-ns/my-topic/my-sub";
private static final String PRODUCE_URI = "ws://localhost:6080/ws/producer/persistent/my-property/cluster1/my-ns/my-topic/";
private static final String CONSUME_URI = "ws://localhost:6080/ws/consumer/persistent/my-property/use/my-ns/my-topic/my-sub";
private static final String PRODUCE_URI = "ws://localhost:6080/ws/producer/persistent/my-property/use/my-ns/my-topic/";
private static final int TEST_PORT = 6080;
private ProxyServer proxyServer;
private WebSocketService service;

@BeforeClass
public void setup() throws Exception {
super.internalSetup();
super.producerBaseSetup();

ServiceConfiguration config = new ServiceConfiguration();
config.setWebServicePort(TEST_PORT);
config.setClusterName("use");
service = spy(new WebSocketService(config));
doReturn(mockZooKeeperClientFactory).when(service).getZooKeeperClientFactory();
service.start();
proxyServer = new ProxyServer(config);
WebSocketServiceStarter.start(proxyServer, service);
log.info("Proxy Server Started");
}

@Override
@AfterClass
protected void cleanup() throws Exception {
super.internalCleanup();
service.close();
proxyServer.stop();
log.info("Finished Cleaning Up Test setup");

}

@Test
public void socketTest() throws InterruptedException {
public void socketTest() throws Exception {
URI consumeUri = URI.create(CONSUME_URI);
URI produceUri = URI.create(PRODUCE_URI);

Expand All @@ -73,19 +82,21 @@ public void socketTest() throws InterruptedException {
try {
consumeClient.start();
ClientUpgradeRequest consumeRequest = new ClientUpgradeRequest();
consumeClient.connect(consumeSocket, consumeUri, consumeRequest);
log.info("Connecting to : %s%n", consumeUri);
Future<Session> consumerFuture = consumeClient.connect(consumeSocket, consumeUri, consumeRequest);
log.info("Connecting to : {}", consumeUri);

ClientUpgradeRequest produceRequest = new ClientUpgradeRequest();
produceClient.start();
produceClient.connect(produceSocket, produceUri, produceRequest);

Future<Session> producerFuture = produceClient.connect(produceSocket, produceUri, produceRequest);
// let it connect
Thread.sleep(1000);
Assert.assertTrue(consumerFuture.get().isOpen());
Assert.assertTrue(producerFuture.get().isOpen());

consumeSocket.awaitClose(1, TimeUnit.SECONDS);
produceSocket.awaitClose(1, TimeUnit.SECONDS);

Assert.assertTrue(produceSocket.getBuffer().size() > 0);
Assert.assertEquals(produceSocket.getBuffer(), consumeSocket.getBuffer());
} catch (Throwable t) {
log.error(t.getMessage());
} finally {
try {
consumeClient.stop();
Expand Down
37 changes: 24 additions & 13 deletions pulsar-broker/src/test/java/com/yahoo/pulsar/websocket/proxy/ProxyPublishConsumeTls.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,60 +22,67 @@
import java.security.KeyManagementException;
import java.security.NoSuchAlgorithmException;
import java.security.SecureRandom;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import javax.net.ssl.KeyManager;
import javax.net.ssl.SSLContext;
import javax.net.ssl.TrustManager;

import org.eclipse.jetty.util.ssl.SslContextFactory;
import org.eclipse.jetty.websocket.api.Session;
import org.eclipse.jetty.websocket.client.ClientUpgradeRequest;
import org.eclipse.jetty.websocket.client.WebSocketClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;

import com.yahoo.pulsar.broker.ServiceConfiguration;
import com.yahoo.pulsar.broker.auth.MockedPulsarServiceBaseTest;
import com.yahoo.pulsar.client.api.ProducerConsumerBase;
import com.yahoo.pulsar.websocket.WebSocketService;
import com.yahoo.pulsar.websocket.service.ProxyServer;
import com.yahoo.pulsar.websocket.service.WebSocketServiceStarter;

import io.netty.handler.ssl.util.InsecureTrustManagerFactory;

public class ProxyPublishConsumeTls extends MockedPulsarServiceBaseTest {
public class ProxyPublishConsumeTls extends ProducerConsumerBase {
protected String methodName;
private static final String CONSUME_URI = "wss://localhost:6090/ws/consumer/persistent/my-property/cluster1/my-ns/my-topic/my-sub";
private static final String PRODUCE_URI = "wss://localhost:6090/ws/producer/persistent/my-property/cluster1/my-ns/my-topic/";
private static final String CONSUME_URI = "wss://localhost:6090/ws/consumer/persistent/my-property/use/my-ns/my-topic/my-sub";
private static final String PRODUCE_URI = "wss://localhost:6090/ws/producer/persistent/my-property/use/my-ns/my-topic/";
private static final String TLS_SERVER_CERT_FILE_PATH = "./src/test/resources/certificate/server.crt";
private static final String TLS_SERVER_KEY_FILE_PATH = "./src/test/resources/certificate/server.key";
private static final int TEST_PORT = 6080;
private static final int TLS_TEST_PORT = 6090;

private ProxyServer proxyServer;
private WebSocketService service;

@BeforeClass
public void setup() throws Exception {
super.internalSetup();
super.producerBaseSetup();

ServiceConfiguration config = new ServiceConfiguration();
config.setWebServicePort(TEST_PORT);
config.setWebServicePortTls(TLS_TEST_PORT);
config.setTlsEnabled(true);
config.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
config.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);

config.setClusterName("use");
service = spy(new WebSocketService(config));
doReturn(mockZooKeeperClientFactory).when(service).getZooKeeperClientFactory();
service.start();

proxyServer = new ProxyServer(config);
WebSocketServiceStarter.start(proxyServer, service);
log.info("Proxy Server Started");
}

@Override
@AfterClass
protected void cleanup() throws Exception {
super.internalCleanup();
service.close();
proxyServer.stop();
log.info("Finished Cleaning Up Test setup");

}
Expand All @@ -101,16 +108,20 @@ public void socketTest() throws InterruptedException, NoSuchAlgorithmException,
try {
consumeClient.start();
ClientUpgradeRequest consumeRequest = new ClientUpgradeRequest();
consumeClient.connect(consumeSocket, consumeUri, consumeRequest);
log.info("Connecting to : %s%n", consumeUri);
Future<Session> consumerFuture = consumeClient.connect(consumeSocket, consumeUri, consumeRequest);
log.info("Connecting to : {}", consumeUri);

ClientUpgradeRequest produceRequest = new ClientUpgradeRequest();
produceClient.start();
produceClient.connect(produceSocket, produceUri, produceRequest);
Future<Session> producerFuture = produceClient.connect(produceSocket, produceUri, produceRequest);
// let it connect
Thread.sleep(1000);
Assert.assertTrue(consumerFuture.get().isOpen());
Assert.assertTrue(producerFuture.get().isOpen());

consumeSocket.awaitClose(1, TimeUnit.SECONDS);
produceSocket.awaitClose(1, TimeUnit.SECONDS);

Assert.assertTrue(produceSocket.getBuffer().size() > 0);
Assert.assertEquals(produceSocket.getBuffer(), consumeSocket.getBuffer());
} catch (Throwable t) {
log.error(t.getMessage());
Expand Down
4 changes: 2 additions & 2 deletions pulsar-broker/src/test/java/com/yahoo/pulsar/websocket/proxy/SimpleConsumerSocket.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,14 +50,14 @@ public boolean awaitClose(int duration, TimeUnit unit) throws InterruptedExcepti

@OnWebSocketClose
public void onClose(int statusCode, String reason) {
log.info("Connection closed: %d - %s%n", statusCode, reason);
log.info("Connection closed: {} - {}", statusCode, reason);
this.session = null;
this.closeLatch.countDown();
}

@OnWebSocketConnect
public void onConnect(Session session) throws InterruptedException {
log.info("Got connect: %s%n", session);
log.info("Got connect: {}", session);
this.session = session;
log.debug("Got connected: {}", session);
}
Expand Down
Loading