Skip to content

Commit fa642e6

Browse files
author
Ink
committed
+ JSON Lines RPC support
1 parent 57ac221 commit fa642e6

File tree

11 files changed

+196
-65
lines changed

11 files changed

+196
-65
lines changed

‎README.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -63,7 +63,7 @@ To enable GPS location passthrough on the Glass you will need to give the applic
6363

6464
Originally, Glass application was serving as a host, and mobile was supposed to connect to it when some updates needed to be passed over, like notification status change or URL intent. But it turned out that the primary use case for the service was to serve as a GPS location provider, since most of my glassware is relying directly on web backends through tethered connection, so I switched the roles. In the future, I can add some 'temporary disconnected' state, when the Glass side will disconnect from the mobile application, but will open listening port so mobile application can 'knock' to re-instantiate the connection.
6565

66-
Uses Java object stream to send data to Google Glass, since I don't want to mess with protocol buffers yet.
66+
Can use Java object stream or JSON Lines to send data to Google Glass, since I don't want to mess with protocol buffers yet.
6767

6868
## AnotherGlass Plans
6969

‎glass-ee/src/main/java/com/damn/anotherglass/glass/ee/host/core/WiFiClient.kt‎

Lines changed: 20 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -10,8 +10,7 @@ import com.damn.anotherglass.shared.rpc.IRPCClient
1010
import com.damn.anotherglass.shared.rpc.RPCHandler
1111
import com.damn.anotherglass.shared.rpc.RPCMessage
1212
import com.damn.anotherglass.shared.rpc.RPCMessageListener
13-
import java.io.ObjectInputStream
14-
import java.io.ObjectOutputStream
13+
import com.damn.anotherglass.shared.rpc.SerializerProvider
1514
import java.net.InetSocketAddress
1615
import java.net.Socket
1716
import java.net.SocketException
@@ -85,26 +84,27 @@ class WiFiClient(private val hostIP: String? = null) : IRPCClient {
8584
}
8685

8786
private fun runLoop(socket: Socket) {
88-
val iss = socket.getInputStream()
89-
val oos = ObjectOutputStream(socket.getOutputStream())
90-
val ois = ObjectInputStream(iss)
91-
while (true) {
92-
while (null != mQueue.peek()) {
93-
val message = mQueue.take()
94-
oos.writeObject(message)
95-
oos.flush()
96-
if (message.service == null) {
97-
return // disconnect requested
87+
socket.getInputStream().use { inputStream ->
88+
socket.getOutputStream().use { outputStream ->
89+
val serializer = SerializerProvider.getSerializer(inputStream, outputStream)
90+
while (true) {
91+
while (null != mQueue.peek()) {
92+
val message = mQueue.take()
93+
serializer.writeMessage(message)
94+
if (message.service == null) {
95+
return // disconnect requested
96+
}
97+
}
98+
while (inputStream.available() > 0) {
99+
val message = serializer.readMessage()
100+
if (null == message.service) {
101+
return
102+
}
103+
mHandler.onDataReceived(message)
104+
}
105+
sleep(100)
98106
}
99107
}
100-
while (iss.available() > 0) {
101-
val message = ois.readObject() as RPCMessage
102-
if (null == message.service) {
103-
return
104-
}
105-
mHandler.onDataReceived(message)
106-
}
107-
sleep(100)
108108
}
109109
}
110110

‎glass-xe/src/main/java/com/damn/anotherglass/glass/host/bluetooth/BluetoothClient.java‎

Lines changed: 9 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -12,16 +12,15 @@
1212

1313
import com.damn.anotherglass.shared.Constants;
1414
import com.damn.anotherglass.shared.rpc.IRPCClient;
15+
import com.damn.anotherglass.shared.rpc.IMessageSerializer;
1516
import com.damn.anotherglass.shared.rpc.RPCHandler;
1617
import com.damn.anotherglass.shared.rpc.RPCMessage;
1718
import com.damn.anotherglass.shared.rpc.RPCMessageListener;
19+
import com.damn.anotherglass.shared.rpc.SerializerProvider;
1820
import com.damn.anotherglass.shared.utility.DisconnectReceiver;
1921
import com.damn.anotherglass.shared.utility.Sleep;
2022

21-
import java.io.IOException;
2223
import java.io.InputStream;
23-
import java.io.ObjectInputStream;
24-
import java.io.ObjectOutputStream;
2524
import java.io.OutputStream;
2625
import java.util.Set;
2726
import java.util.concurrent.BlockingQueue;
@@ -97,31 +96,29 @@ public void shutdown() {
9796
}
9897

9998
@SuppressLint("MissingPermission")
100-
private void runLoop(@NonNull BluetoothDevice device) throws IOException, InterruptedException, ClassNotFoundException {
99+
private void runLoop(@NonNull BluetoothDevice device) throws Exception {
101100
try (BluetoothSocket socket = device.createInsecureRfcommSocketToServiceRecord(Constants.uuid)) {
102101
socket.connect();
103102
Log.i(TAG, "Client has connected to " + device.getName());
104103
AtomicBoolean active = new AtomicBoolean(true);
105104
try (DisconnectReceiver ignored = new DisconnectReceiver(mContext, device, () -> active.getAndSet(false))) {
106105
try (OutputStream outputStream = socket.getOutputStream();
107-
InputStream inputStream = socket.getInputStream()) {
108-
ObjectOutputStream os = new ObjectOutputStream(outputStream);
109-
ObjectInputStream in = new ObjectInputStream(inputStream);
106+
InputStream inputStream = socket.getInputStream()) {
107+
IMessageSerializer serializer = SerializerProvider.getSerializer(inputStream, outputStream);
110108
mConnected = true;
111109
mHandler.onConnectionStarted(device.getName());
112110
while (active.get()) {
113-
while(null != mQueue.peek()){
111+
while (null != mQueue.peek()) {
114112
RPCMessage message = mQueue.take();
115-
os.writeObject(message);
113+
serializer.writeMessage(message);
116114
Log.v(TAG, "Message " + message.service + "/" + message.type + " was sent");
117-
if(null == message.service) {
115+
if (null == message.service) {
118116
Log.d(TAG, "Shutdown requested");
119-
os.flush();
120117
return;
121118
}
122119
}
123120
while (inputStream.available() > 0) {
124-
RPCMessage objectReceived = (RPCMessage) in.readObject();
121+
RPCMessage objectReceived = serializer.readMessage();
125122
mHandler.onDataReceived(objectReceived);
126123
Log.v(TAG, "Message " + objectReceived.service + "/" + objectReceived.type + " was received");
127124
}

‎mobile/src/main/java/com/damn/anotherglass/core/BluetoothHost.java‎

Lines changed: 8 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -10,23 +10,22 @@
1010
import android.content.pm.PackageManager;
1111
import android.util.Log;
1212

13-
import androidx.annotation.CallSuper;
1413
import androidx.core.app.ActivityCompat;
1514

1615
import com.applicaster.xray.android.adapters.ALog;
1716
import com.damn.anotherglass.shared.Constants;
1817
import com.damn.anotherglass.shared.rpc.IRPCHost;
18+
import com.damn.anotherglass.shared.rpc.IMessageSerializer;
1919
import com.damn.anotherglass.shared.rpc.RPCHandler;
2020
import com.damn.anotherglass.shared.rpc.RPCMessage;
2121
import com.damn.anotherglass.shared.rpc.RPCMessageListener;
22+
import com.damn.anotherglass.shared.rpc.SerializerProvider;
2223
import com.damn.anotherglass.shared.utility.Closeables;
2324
import com.damn.anotherglass.shared.utility.DisconnectReceiver;
2425
import com.damn.anotherglass.shared.utility.Sleep;
2526

2627
import java.io.IOException;
2728
import java.io.InputStream;
28-
import java.io.ObjectInputStream;
29-
import java.io.ObjectOutputStream;
3029
import java.io.OutputStream;
3130
import java.util.concurrent.BlockingQueue;
3231
import java.util.concurrent.LinkedBlockingDeque;
@@ -112,7 +111,7 @@ public void run() {
112111
if (socket == null)
113112
continue;
114113
runLoop(socket);
115-
} catch (IOException | ClassNotFoundException | InterruptedException e) {
114+
} catch (Exception e) {
116115
Log.e(TAG, "Exception in runLoop: " + e, e);
117116
mHandler.onConnectionLost(e.getLocalizedMessage());
118117
} finally {
@@ -143,25 +142,25 @@ public void shutdown() {
143142
}
144143

145144
@SuppressLint("MissingPermission")
146-
private void runLoop(BluetoothSocket socket) throws IOException, ClassNotFoundException, InterruptedException {
145+
private void runLoop(BluetoothSocket socket) throws Exception {
147146
final BluetoothDevice remoteDevice = socket.getRemoteDevice();
148147
ALog.d(TAG, "Connected to " + remoteDevice.getName());
149148
mHandler.onConnectionStarted(remoteDevice.getName());
150149
try (DisconnectReceiver ignored = new DisconnectReceiver(mContext, remoteDevice, this::onConnectionLost)) {
150+
151151
try (InputStream inputStream = socket.getInputStream();
152152
OutputStream outputStream = socket.getOutputStream()) {
153-
ObjectInputStream in = new ObjectInputStream(inputStream);
154-
ObjectOutputStream os = new ObjectOutputStream(outputStream);
153+
IMessageSerializer serializer = SerializerProvider.getSerializer(inputStream, outputStream);
155154
while (mActive) {
156155
while (inputStream.available() > 0) {
157-
RPCMessage objectReceived = (RPCMessage) in.readObject();
156+
RPCMessage objectReceived = serializer.readMessage();
158157
if (null == objectReceived.service)
159158
return; // shutdown requested
160159
mHandler.onDataReceived(objectReceived);
161160
}
162161
while (null != mQueue.peek()) {
163162
RPCMessage message = mQueue.take();
164-
os.writeObject(message);
163+
serializer.writeMessage(message);
165164
}
166165
Sleep.sleep(100);
167166
}

‎mobile/src/main/java/com/damn/anotherglass/core/WiFiHost.kt‎

Lines changed: 21 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -8,10 +8,9 @@ import com.damn.anotherglass.shared.rpc.IRPCHost
88
import com.damn.anotherglass.shared.rpc.RPCHandler
99
import com.damn.anotherglass.shared.rpc.RPCMessage
1010
import com.damn.anotherglass.shared.rpc.RPCMessageListener
11+
import com.damn.anotherglass.shared.rpc.SerializerProvider
1112
import com.damn.anotherglass.shared.utility.Closeables
1213
import com.damn.anotherglass.shared.utility.Sleep
13-
import java.io.ObjectInputStream
14-
import java.io.ObjectOutputStream
1514
import java.net.ServerSocket
1615
import java.net.Socket
1716
import java.net.SocketException
@@ -38,7 +37,7 @@ class WiFiHost(listener: RPCMessageListener) : IRPCHost {
3837
}
3938

4039
override fun send(message: RPCMessage) {
41-
when(val workerThread = mWorkerThread) {
40+
when (val workerThread = mWorkerThread) {
4241
null -> logger.e(TAG, "Not started")
4342
else -> if (!workerThread.isConnected())
4443
logger.e(TAG, "Not connected")
@@ -100,26 +99,27 @@ class WiFiHost(listener: RPCMessageListener) : IRPCHost {
10099
}
101100

102101
private fun runLoop(socket: Socket) {
103-
val iss = socket.getInputStream()
104-
val oos = ObjectOutputStream(socket.getOutputStream())
105-
val ois = ObjectInputStream(iss)
106-
while (mActive) {
107-
while (mQueue.peek() != null) {
108-
val message = mQueue.take()
109-
oos.writeObject(message)
110-
oos.flush()
111-
if (message.service == null) {
112-
return // disconnect requested
113-
}
114-
}
115-
while (mActive && iss.available() > 0) {
116-
val message = ois.readObject() as RPCMessage
117-
if (message.service == null) {
118-
return // client disconnected
102+
socket.getInputStream().use { inputStream ->
103+
socket.getOutputStream().use { outputStream ->
104+
val serializer = SerializerProvider.getSerializer(inputStream, outputStream)
105+
while (mActive) {
106+
while (mQueue.peek() != null) {
107+
val message = mQueue.take()
108+
serializer.writeMessage(message)
109+
if (message.service == null) {
110+
return // disconnect requested
111+
}
112+
}
113+
while (mActive && inputStream.available() > 0) {
114+
val message = serializer.readMessage()
115+
if (message.service == null) {
116+
return // client disconnected
117+
}
118+
mHandler.onDataReceived(message)
119+
}
120+
Sleep.sleep(100)
119121
}
120-
mHandler.onDataReceived(message)
121122
}
122-
Sleep.sleep(100)
123123
}
124124
}
125125

‎shared/build.gradle‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,4 +20,5 @@ android {
2020

2121
dependencies {
2222
implementation "androidx.annotation:annotation:1.9.1"
23+
implementation "com.google.code.gson:gson:2.13.1"
2324
}

‎shared/src/main/java/com/damn/anotherglass/shared/rpc/IMessageSerializer.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
package com.damn.anotherglass.shared.rpc;
2+
3+
public interface IMessageSerializer {
4+
void writeMessage(RPCMessage message) throws Exception;
5+
RPCMessage readMessage() throws Exception;
6+
}

‎shared/src/main/java/com/damn/anotherglass/shared/rpc/JsonMessageSerializer.java‎

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,73 @@
1+
package com.damn.anotherglass.shared.rpc;
2+
3+
import com.google.gson.Gson;
4+
import com.google.gson.GsonBuilder;
5+
import com.google.gson.JsonDeserializationContext;
6+
import com.google.gson.JsonDeserializer;
7+
import com.google.gson.JsonElement;
8+
import com.google.gson.JsonObject;
9+
import com.google.gson.JsonParseException;
10+
import com.google.gson.Strictness;
11+
12+
import java.io.BufferedReader;
13+
import java.io.InputStream;
14+
import java.io.InputStreamReader;
15+
import java.io.OutputStream;
16+
import java.io.OutputStreamWriter;
17+
import java.lang.reflect.Type;
18+
import java.nio.charset.StandardCharsets;
19+
20+
// Basic JSON Lines message serializer
21+
class JsonMessageSerializer implements IMessageSerializer {
22+
private final Gson gson;
23+
private final OutputStreamWriter writer;
24+
private final BufferedReader reader;
25+
26+
public JsonMessageSerializer(InputStream inputStream, OutputStream outputStream) {
27+
gson = new GsonBuilder()
28+
.registerTypeAdapter(RPCMessage.class, new RPCMessageDeserializer())
29+
.setStrictness(Strictness.LENIENT)
30+
.create();
31+
writer = new OutputStreamWriter(outputStream, StandardCharsets.UTF_8);
32+
reader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8));
33+
}
34+
35+
@Override
36+
public void writeMessage(RPCMessage message) throws Exception {
37+
gson.toJson(message, writer);
38+
writer.write("\n"); // maybe use Record Separator code?
39+
writer.flush();
40+
}
41+
42+
@Override
43+
public RPCMessage readMessage() throws Exception {
44+
// do not read from the stream directly,
45+
// since it will looks like multiple concatenated jsons
46+
String line = reader.readLine();
47+
return gson.fromJson(line, RPCMessage.class);
48+
}
49+
50+
private static class RPCMessageDeserializer implements JsonDeserializer<RPCMessage> {
51+
@Override
52+
public RPCMessage deserialize(JsonElement json, Type typeOfT, JsonDeserializationContext context) throws JsonParseException {
53+
JsonObject jsonObject = json.getAsJsonObject();
54+
55+
String service = jsonObject.get("service").getAsString();
56+
String type = null;
57+
if (jsonObject.has("type") && !jsonObject.get("type").isJsonNull()) {
58+
type = jsonObject.get("type").getAsString();
59+
}
60+
61+
Object payload = null;
62+
if (type != null && jsonObject.has("payload") && !jsonObject.get("payload").isJsonNull()) {
63+
try {
64+
Class<?> payloadClass = Class.forName(type); // todo: cache class lookup
65+
payload = context.deserialize(jsonObject.get("payload"), payloadClass);
66+
} catch (ClassNotFoundException e) {
67+
throw new JsonParseException("Unable to find class: " + type + " for RPCMessage payload", e);
68+
}
69+
}
70+
return new RPCMessage(service, type, payload);
71+
}
72+
}
73+
}

‎shared/src/main/java/com/damn/anotherglass/shared/rpc/ObjectMessageSerializer.java‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
1+
package com.damn.anotherglass.shared.rpc;
2+
3+
import java.io.IOException;
4+
import java.io.ObjectInputStream;
5+
import java.io.ObjectOutputStream;
6+
import java.io.InputStream;
7+
import java.io.OutputStream;
8+
9+
class ObjectMessageSerializer implements IMessageSerializer {
10+
11+
private final ObjectInputStream ois;
12+
private final ObjectOutputStream oos;
13+
14+
ObjectMessageSerializer(InputStream inputStream, OutputStream outputStream) throws IOException {
15+
ois = new ObjectInputStream(inputStream);
16+
oos = new ObjectOutputStream(outputStream);
17+
}
18+
19+
@Override
20+
public void writeMessage(RPCMessage message) throws Exception {
21+
oos.writeObject(message);
22+
oos.flush();
23+
}
24+
25+
@Override
26+
public RPCMessage readMessage() throws Exception {
27+
return (RPCMessage) ois.readObject();
28+
}
29+
}

‎shared/src/main/java/com/damn/anotherglass/shared/rpc/RPCMessage.java‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,13 @@ public class RPCMessage implements Serializable {
99

1010
public <T extends Serializable> RPCMessage(String service, T obj) {
1111
this.service = service;
12-
type = null != obj ? obj.getClass().getName() : null;
13-
payload = obj;
12+
this.type = null != obj ? obj.getClass().getName() : null;
13+
this.payload = obj;
14+
}
15+
16+
public RPCMessage(String service, String typeName, Object payloadObj) {
17+
this.service = service;
18+
this.type = typeName;
19+
this.payload = payloadObj;
1420
}
1521
}

0 commit comments

Comments
 (0)