Skip to content

Commit 4031a1c

Browse files
jpwsuttonjpwsutton
committed
Merging New Websocket Header enhancement with Updated Network Module enhancement
Signed-off-by: James Sutton <james.sutton@uk.ibm.com>
2 parents 6d2906e + d669d22 commit 4031a1c

17 files changed

Lines changed: 123 additions & 27 deletions

File tree

‎MQTTv3.md‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -98,3 +98,20 @@ public class MqttPublishSample {
9898
}
9999
```
100100

101+
## Adding custom headers for Websocket connection
102+
103+
The included code below is a extended basic sample that connects to a server with custom headers.
104+
105+
```
106+
MqttClient client = new MqttClient("wss://<BROKER_URI>", "MyClient");
107+
108+
MqttConnectOptions connectOptions = new MqttConnectOptions();
109+
Properties properties = new Properties();
110+
properties.setProperty("X-Amz-CustomAuthorizer-Name", <SOME_VALUE>);
111+
properties.setProperty("X-Amz-CustomAuthorizer-Signature", <SOME_VALUE>);
112+
properties.setProperty(<SOME_VALUE>, <SOME_VALUE>);
113+
connectOptions.setCustomWebSocketHeaders(properties);
114+
115+
client.connect(connectOptions);
116+
117+
```

org.eclipse.paho.client.mqttv3.test/src/test/java/org/eclipse/paho/client/mqttv3/internal/NetworkModuleServiceTest.java renamed to org.eclipse.paho.client.mqttv3.test/src/test/java/org/eclipse/paho/client/mqttv3/test/NetworkModuleServiceTest.java

Lines changed: 9 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,17 +1,20 @@
1-
package org.eclipse.paho.client.mqttv3.internal;
1+
package org.eclipse.paho.client.mqttv3.test;
2+
3+
import static org.junit.Assert.assertEquals;
4+
import static org.junit.Assert.assertNull;
5+
import static org.junit.Assert.assertTrue;
6+
import static org.junit.Assert.fail;
27

38
import java.net.URI;
49
import java.net.URISyntaxException;
510

611
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
712
import org.eclipse.paho.client.mqttv3.MqttException;
13+
import org.eclipse.paho.client.mqttv3.internal.NetworkModule;
14+
import org.eclipse.paho.client.mqttv3.internal.NetworkModuleService;
15+
import org.eclipse.paho.client.mqttv3.internal.TCPNetworkModule;
816
import org.junit.Test;
917

10-
import static org.junit.Assert.assertEquals;
11-
import static org.junit.Assert.assertNull;
12-
import static org.junit.Assert.assertTrue;
13-
import static org.junit.Assert.fail;
14-
1518
public class NetworkModuleServiceTest {
1619

1720
@Test

‎org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/MqttAsyncClient.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -538,7 +538,9 @@ private NetworkModule createNetworkModule(String address, MqttConnectOptions opt
538538
// @TRACE 115=URI={0}
539539
log.fine(CLASS_NAME,methodName, "115", new Object[] {address});
540540

541+
541542
NetworkModule netModule = NetworkModuleService.createInstance(address, options, clientId);
543+
542544
return netModule;
543545
}
544546

‎org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/MqttConnectOptions.java‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,7 @@ public class MqttConnectOptions {
7373
private int mqttVersion = MQTT_VERSION_DEFAULT;
7474
private boolean automaticReconnect = false;
7575
private int maxReconnectDelay = 128000;
76+
private Properties customWebSocketHeaders = null;
7677

7778
/**
7879
* Constructs a new <code>MqttConnectOptions</code> object using the
@@ -602,6 +603,20 @@ public Properties getDebug() {
602603
return p;
603604
}
604605

606+
/**
607+
* Sets the Custom WebSocket Headers for the WebSocket Connection.
608+
*
609+
* @param props The custom websocket headers {@link Properties}
610+
*/
611+
612+
public void setCustomWebSocketHeaders(Properties props) {
613+
this.customWebSocketHeaders = props;
614+
}
615+
616+
public Properties getCustomWebSocketHeaders() {
617+
return customWebSocketHeaders;
618+
}
619+
605620
public String toString() {
606621
return Debug.dumpProperties(getDebug(), "Connection options");
607622
}

‎org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/internal/NetworkModuleService.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -115,10 +115,10 @@ public static NetworkModule createInstance(String address, MqttConnectOptions op
115115
* This workaround tries to detect such a parsing failure and does tokenize the authority parts according to
116116
* RFC3986, but does not enforce any character restrictions (for sake of simplicity).
117117
*
118-
* @param toPatch
119-
* @see https://tools.ietf.org/html/rfc3986#section-3.2
118+
* @param toPatch - The URI To patch
119+
* @see <a href="https://tools.ietf.org/html/rfc3986#section-3.2">rfc3986 - section-3.2</a>
120120
*/
121-
static void applyRFC3986AuthorityPatch(URI toPatch) {
121+
public static void applyRFC3986AuthorityPatch(URI toPatch) {
122122
if (toPatch == null
123123
|| toPatch.getHost() != null // already successfully parsed
124124
|| toPatch.getAuthority() == null

‎org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/internal/websocket/WebSocketHandshake.java‎

Lines changed: 16 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,9 @@
2929
import java.util.HashMap;
3030
import java.util.Map;
3131
import java.util.UUID;
32+
import java.util.Properties;
33+
import java.util.Set;
34+
import java.util.Iterator;
3235
/**
3336
* Helper class to execute a WebSocket Handshake.
3437
*/
@@ -52,14 +55,15 @@ public class WebSocketHandshake {
5255
String uri;
5356
String host;
5457
int port;
58+
Properties customWebSocketHeaders;
5559

56-
57-
public WebSocketHandshake(InputStream input, OutputStream output, String uri, String host, int port){
60+
public WebSocketHandshake(InputStream input, OutputStream output, String uri, String host, int port, Properties customWebSocketHeaders){
5861
this.input = input;
5962
this.output = output;
6063
this.uri = uri;
6164
this.host = host;
6265
this.port = port;
66+
this.customWebSocketHeaders = customWebSocketHeaders;
6367
}
6468

6569

@@ -108,6 +112,16 @@ private void sendHandshakeRequest(String key) throws IOException{
108112
pw.print("Sec-WebSocket-Protocol: mqtt" + LINE_SEPARATOR);
109113
pw.print("Sec-WebSocket-Version: 13" + LINE_SEPARATOR);
110114

115+
if (customWebSocketHeaders != null) {
116+
Set keys = customWebSocketHeaders.keySet();
117+
Iterator i = keys.iterator();
118+
while (i.hasNext()) {
119+
String k = (String) i.next();
120+
String value = customWebSocketHeaders.getProperty(k);
121+
pw.print(k + ": " + value + LINE_SEPARATOR);
122+
}
123+
}
124+
111125
String userInfo = srvUri.getUserInfo();
112126
if(userInfo != null) {
113127
pw.print("Authorization: Basic " + Base64.encode(userInfo) + LINE_SEPARATOR);

‎org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/internal/websocket/WebSocketNetworkModule.java‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
import java.io.OutputStream;
2222
import java.io.PipedInputStream;
2323
import java.nio.ByteBuffer;
24+
import java.util.Properties;
2425

2526
import javax.net.SocketFactory;
2627

@@ -37,6 +38,7 @@ public class WebSocketNetworkModule extends TCPNetworkModule {
3738
private String uri;
3839
private String host;
3940
private int port;
41+
private Properties customWebsocketHeaders;
4042
private PipedInputStream pipedInputStream;
4143
private WebSocketReceiver webSocketReceiver;
4244
ByteBuffer recievedPayload;
@@ -47,20 +49,21 @@ public class WebSocketNetworkModule extends TCPNetworkModule {
4749
* Frame before passing it through to the real socket.
4850
*/
4951
private ByteArrayOutputStream outputStream = new ExtendedByteArrayOutputStream(this);
50-
51-
public WebSocketNetworkModule(SocketFactory factory, String uri, String host, int port, String resourceContext){
52+
53+
public WebSocketNetworkModule(SocketFactory factory, String uri, String host, int port, String resourceContext, Properties customWebsocketHeaders){
5254
super(factory, host, port, resourceContext);
5355
this.uri = uri;
5456
this.host = host;
5557
this.port = port;
58+
this.customWebsocketHeaders = customWebsocketHeaders;
5659
this.pipedInputStream = new PipedInputStream();
5760

5861
log.setResourceName(resourceContext);
5962
}
6063

6164
public void start() throws IOException, MqttException {
6265
super.start();
63-
WebSocketHandshake handshake = new WebSocketHandshake(getSocketInputStream(), getSocketOutputStream(), uri, host, port);
66+
WebSocketHandshake handshake = new WebSocketHandshake(getSocketInputStream(), getSocketOutputStream(), uri, host, port, customWebsocketHeaders);
6467
handshake.execute();
6568
this.webSocketReceiver = new WebSocketReceiver(getSocketInputStream(), pipedInputStream);
6669
webSocketReceiver.start("webSocketReceiver");

‎org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/internal/websocket/WebSocketNetworkModuleFactory.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -52,7 +52,7 @@ public NetworkModule createNetworkModule(URI brokerUri, MqttConnectOptions optio
5252
throw ExceptionHelper.createMqttException(MqttException.REASON_CODE_SOCKET_FACTORY_MISMATCH);
5353
}
5454
WebSocketNetworkModule netModule = new WebSocketNetworkModule(factory, brokerUri.toString(), host, port,
55-
clientId);
55+
clientId, options.getCustomWebSocketHeaders());
5656
netModule.setConnectTimeout(options.getConnectionTimeout());
5757
return netModule;
5858
}

‎org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/internal/websocket/WebSocketSecureNetworkModule.java‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@
2121
import java.io.OutputStream;
2222
import java.io.PipedInputStream;
2323
import java.nio.ByteBuffer;
24-
24+
import java.util.Properties;
2525
import javax.net.ssl.SSLSocketFactory;
2626

2727
import org.eclipse.paho.client.mqttv3.MqttException;
@@ -39,6 +39,7 @@ public class WebSocketSecureNetworkModule extends SSLNetworkModule{
3939
private String uri;
4040
private String host;
4141
private int port;
42+
private Properties customWebSocketHeaders;
4243
ByteBuffer recievedPayload;
4344

4445
/**
@@ -48,18 +49,19 @@ public class WebSocketSecureNetworkModule extends SSLNetworkModule{
4849
*/
4950
private ByteArrayOutputStream outputStream = new ExtendedByteArrayOutputStream(this);
5051

51-
public WebSocketSecureNetworkModule(SSLSocketFactory factory, String uri, String host, int port, String clientId) {
52+
public WebSocketSecureNetworkModule(SSLSocketFactory factory, String uri, String host, int port, String clientId, Properties customWebSocketHeaders) {
5253
super(factory, host, port, clientId);
5354
this.uri = uri;
5455
this.host = host;
5556
this.port = port;
57+
this.customWebSocketHeaders = customWebSocketHeaders;
5658
this.pipedInputStream = new PipedInputStream();
5759
log.setResourceName(clientId);
5860
}
5961

6062
public void start() throws IOException, MqttException {
6163
super.start();
62-
WebSocketHandshake handshake = new WebSocketHandshake(super.getInputStream(), super.getOutputStream(), uri, host, port);
64+
WebSocketHandshake handshake = new WebSocketHandshake(super.getInputStream(), super.getOutputStream(), uri, host, port, customWebSocketHeaders);
6365
handshake.execute();
6466
this.webSocketReceiver = new WebSocketReceiver(getSocketInputStream(), pipedInputStream);
6567
webSocketReceiver.start("WssSocketReceiver");

‎org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/internal/websocket/WebSocketSecureNetworkModuleFactory.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -64,7 +64,7 @@ public NetworkModule createNetworkModule(URI brokerUri, MqttConnectOptions optio
6464

6565
// Create the network module...
6666
WebSocketSecureNetworkModule netModule = new WebSocketSecureNetworkModule((SSLSocketFactory) factory,
67-
brokerUri.toString(), host, port, clientId);
67+
brokerUri.toString(), host, port, clientId, options.getCustomWebSocketHeaders());
6868
netModule.setSSLhandshakeTimeout(options.getConnectionTimeout());
6969
netModule.setSSLHostnameVerifier(options.getSSLHostnameVerifier());
7070
netModule.setHttpsHostnameVerificationEnabled(options.isHttpsHostnameVerificationEnabled());

0 commit comments

Comments
 (0)