diff --git a/add-on/background.js b/add-on/background.js index bcbb69a..5142b5c 100644 --- a/add-on/background.js +++ b/add-on/background.js @@ -4,10 +4,42 @@ On startup, connect to the "ping_pong" app. */ var port = browser.runtime.connectNative("universalmusic"); +var listeners = [function (message, returnValue) { + if (returnValue == null) + { + returnValue = "pong"; + } + return returnValue; +}]; + +function sleep(ms) { + return new Promise(resolve => setTimeout(resolve, ms)); +} + +function handleMessage(message) +{ + return new Promise((resolve, reject) => { + var returnValue = null; + for (index in listeners) + { + returnValue = listeners[index](message, returnValue); + } + resolve(returnValue); + }) +} + /* -Listen for messages from the app. -*/ -port.onMessage.addListener((response) => { - console.log("Received: " + response); - port.postMessage("pong"); + * Listen for messages from the app. + */ +port.onMessage.addListener((message) => { + console.log("Received: ", message); + + handleMessage(message.message).then((response) => { + returnValue = { + "messageNum": message.messageNum, + "message": response + } + console.log("Sending ", returnValue) + port.postMessage(returnValue); + }); }); diff --git a/addonInter/build.gradle b/addonInter/build.gradle index fe5372b..7649933 100644 --- a/addonInter/build.gradle +++ b/addonInter/build.gradle @@ -31,6 +31,8 @@ dependencies { implementation 'com.google.code.gson:gson:2.8.7' + implementation project(":browserCommands") + // Declare the dependency for your favourite test framework you want to use in your tests. // TestNG is also supported by the Gradle Test task. Just change the // testCompile dependency to testCompile 'org.testng:testng:6.8.1' and add diff --git a/addonInter/src/main/java/edu/regis/universeplayer/addon/BrowserLink.java b/addonInter/src/main/java/edu/regis/universeplayer/addon/BrowserLink.java index 548368e..360cbb3 100644 --- a/addonInter/src/main/java/edu/regis/universeplayer/addon/BrowserLink.java +++ b/addonInter/src/main/java/edu/regis/universeplayer/addon/BrowserLink.java @@ -5,21 +5,17 @@ package edu.regis.universeplayer.addon; import com.google.gson.*; +import edu.regis.universeplayer.browserCommands.MessageRunner; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.BufferedInputStream; -import java.io.BufferedOutputStream; import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; import java.nio.ByteBuffer; import java.nio.ByteOrder; import java.nio.charset.StandardCharsets; import java.util.HashMap; -import java.util.LinkedList; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.Future; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; /** * The browser link serves as a communication between this process and the @@ -29,342 +25,98 @@ import java.util.concurrent.TimeoutException; * @author William Hubbard * @version 0.1 */ -public class BrowserLink implements Runnable +public class BrowserLink extends MessageRunner { - private static Logger logger = LoggerFactory.getLogger(BrowserLink.class); + private static final Logger logger = LoggerFactory.getLogger(BrowserLink.class); public static final Gson gson = new Gson(); - private BufferedInputStream browserIn; - private BufferedOutputStream browserOut; - - private final Object readLock = new Object(); - /** - * This queue serves as a cache for objects we need to send. + * Creates a message runner. */ - private final LinkedList sendQueue = new LinkedList<>(); - /** - * This queue serves as a cache for objects that are waiting for a response. - */ - private final LinkedList sentQueue = new LinkedList<>(); - - private boolean running; + public BrowserLink() + { + super(System.in, System.out); + } @Override - public void run() + public byte[] serializeObject(Object message) { - MessagePacket packet; - this.running = true; - try - { - browserIn = new BufferedInputStream(System.in); - browserOut = new BufferedOutputStream(System.out); - - while (this.running) - { - /* - * Sends all messages - */ - synchronized (this.sendQueue) - { - packet = this.sendQueue.poll(); - } - if (packet != null) - { - try - { - logger.debug("Writing message {} {}", packet.message[0], packet.message[1]); - this.browserOut.write(packet.message[0]); - this.browserOut.write(packet.message[1]); - this.browserOut.flush(); - - synchronized (this.sentQueue) - { - this.sentQueue.add(packet); - } - } - catch (IOException e) - { - logger.error("Could not send message " + new String(packet.message[1], StandardCharsets.UTF_8), e); - } - } - - try - { - packet = null; - synchronized (this.sentQueue) - { - if (this.sentQueue.size() > 0) - { - packet = this.sentQueue.poll(); - } - } - if (packet != null) - { - if (this.browserIn.available() > 0) - { - /* - * Wait for a response from the browser. - */ - byte[][] returnMessage = this.readMessage(); - if (returnMessage == null) - { - logger.info("Browser connection closed."); - this.running = false; - } - else - { - logger.debug("Reading message {} {}", returnMessage[0], returnMessage[1]); - packet.returnMessage = returnMessage; - synchronized (this.readLock) - { - logger.trace("Received message, notifying futures."); - this.readLock.notifyAll(); - } - } - } - else - { - synchronized (this.sentQueue) - { - this.sentQueue.addFirst(packet); - } - logger.trace("There are no new messages"); - } - } - } - catch (IOException e) - { - logger.error("Could not retrieve message", e); - } - } - } - finally - { - /* - * Release locks - */ - synchronized (this.sendQueue) - { - while (this.sendQueue.size() > 0) - { - packet = this.sendQueue.poll(); - packet.returnMessage = null; - } - } - synchronized (this.readLock) - { - this.readLock.notifyAll(); - } - try - { - browserIn.close(); - } - catch (IOException e1) - { - logger.error("Could not close browser input", e1); - } - finally - { - try - { - browserOut.close(); - } - catch (IOException e1) - { - logger.error("Could not close browser output", e1); - } - } - } + String val = gson.toJson(message); + return val.getBytes(StandardCharsets.UTF_8); } - /** - * Reads a message from the browser input. - * - * @return - A new message packet from the browser, or null if it has been closed. - * @throws IOException - Should a read error occur. - */ - private byte[][] readMessage() throws IOException + @Override + public Object deserializeObject(byte[] message) throws IOException { - ByteBuffer headerReader; - int rawLength, messageLength; - byte[] header = new byte[4]; - byte[] message; - - rawLength = browserIn.read(header); + JsonElement val = gson.fromJson(new String(message, StandardCharsets.UTF_8), JsonElement.class); + return getMessage(val); + } + + @Override + public void writeMessage(OutputStream out, int messageNum, byte[] message) throws IOException + { + String messageString; + JsonObject sendOb = new JsonObject(); + sendOb.add("messageNum", new JsonPrimitive(messageNum)); + sendOb.add("message", gson.fromJson(new String(message, StandardCharsets.UTF_8), JsonElement.class)); + messageString = gson.toJson(sendOb); + message = messageString.getBytes(StandardCharsets.UTF_8); + ByteBuffer lengthBuffer = ByteBuffer.allocate(4); + lengthBuffer.order(ByteOrder.nativeOrder()); + logger.debug("Writing message {} {}", messageNum, messageString); /* - * A value of 0 or -1 indicates that there are no more messages, and - * that the channel has been closed. + * Writes the message length */ - if (rawLength <= 0) + lengthBuffer.clear(); + lengthBuffer.putInt(message.length); + out.write(lengthBuffer.array()); + /* + * Writes the message + */ + out.write(message); + out.flush(); + } + + @Override + public byte[][] readMessage(InputStream in) throws IOException + { + int messageLength; + int messageNum; + int readLength; + byte[] message; + JsonObject messageJson; + ByteBuffer lengthBuffer = ByteBuffer.allocate(4); + lengthBuffer.order(ByteOrder.nativeOrder()); + lengthBuffer.clear(); + readLength = in.read(lengthBuffer.array()); + if (readLength == 0) { - logger.info("The browser channel has been closed."); + logger.debug("Input stream closed, no more messages."); return null; } - - headerReader = ByteBuffer.wrap(header, 0, rawLength); - /* - * The browser outputs in whatever the native endian order, - * which may not be what Java does. - */ - headerReader.order(ByteOrder.nativeOrder()); - messageLength = headerReader.getInt(); - logger.debug("Receiving message of {} length", messageLength); + messageLength = lengthBuffer.getInt(); message = new byte[messageLength]; - browserIn.read(message); - logger.debug("Unpacking message {} (length {})", new String(message, StandardCharsets.UTF_8), messageLength); - return new byte[][] {header, message}; - } - - /** - * Sends an object to the browser. - * - * @param message - The object to send. - * @return A future for the response from the browser. - */ - public Future sendObject(Object message) - { - String messageJson; - byte[] messageData; - ByteBuffer header; - byte[][] packet; - Future future; - MessagePacket packetEntry; - logger.debug("Sending message {}", message); - - messageJson = gson.toJson(message); - messageData = messageJson.getBytes(StandardCharsets.UTF_8); - header = ByteBuffer.allocate(4); - /* - * The browser outputs in whatever the native endian order is, - * which may not be what Java does. - */ - header.order(ByteOrder.nativeOrder()); - header.putInt(messageData.length); - - packetEntry = new MessagePacket(new byte[][] {header.array(), messageData}); - - future = new Future<>() + readLength = in.read(message); + if (readLength < message.length) { - private boolean canceled = false; - - @Override - public boolean cancel(boolean mayInterruptIfRunning) - { - synchronized (sendQueue) - { - if (sendQueue.remove(packetEntry)) - { - canceled = true; - if (mayInterruptIfRunning) - { - logger.trace("Canceling future, notifying self futures."); - synchronized (readLock) - { - readLock.notifyAll(); - } - } - else - { - logger.trace("Canceling future."); - } - return true; - } - else - { - return false; - } - } - } - - @Override - public boolean isCancelled() - { - return this.canceled; - } - - @Override - public boolean isDone() - { - return packetEntry.returnMessage != null; - } - - @Override - public Object get() throws InterruptedException, ExecutionException - { - logger.trace("Future waiting for message"); - synchronized (readLock) - { - while (!this.canceled && packetEntry.returnMessage == null) - { - readLock.wait(); - } - } - if (packetEntry.returnMessage == null) - { - throw new InterruptedException(); - } - logger.trace("Future has received message"); - try - { - return getMessage(packetEntry.returnMessage[1]); - } - catch (IOException e) - { - throw new ExecutionException(e); - } - } - - @Override - public Object get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException - { - logger.trace("Future waiting for message"); - synchronized (readLock) - { - while (!this.canceled && packetEntry.returnMessage == null) - { - readLock.wait(unit.toMillis(timeout)); - } - } - if (packetEntry.returnMessage == null) - { - throw new InterruptedException(); - } - logger.trace("Future has received message"); - try - { - return getMessage(packetEntry.returnMessage[1]); - } - catch (IOException e) - { - throw new ExecutionException(e); - } - } - }; - - packetEntry.returnValue = future; - - synchronized (this.sendQueue) - { - sendQueue.add(packetEntry); + logger.warn("Message shorter than reported (expected {} bytes, got {} bytes)", message.length, readLength); } + messageJson = gson.fromJson(new String(message, StandardCharsets.UTF_8), JsonObject.class); + messageNum = messageJson.get("messageNum").getAsInt(); + message = gson.toJson(messageJson.get("message")).getBytes(StandardCharsets.UTF_8); - return future; - } - - /** - * Queries for the next message from the browser. - * - * @return The received message - * @throws IOException - If there is an error when reading the next object, - * or if the browser connection is closed. - */ - private Object getMessage(byte[] messageData) throws IOException - { - JsonElement message = gson.fromJson(new String(messageData, StandardCharsets.UTF_8), JsonElement.class); - return getMessage(message); + lengthBuffer.order(ByteOrder.BIG_ENDIAN); + /* + * Wipe out the buffer + */ + lengthBuffer.clear(); + lengthBuffer.put(new byte[4]); + lengthBuffer.clear(); + lengthBuffer.putInt(messageNum); + logger.debug("Reading message {}", messageNum); + return new byte[][] {lengthBuffer.array(), message}; } /** @@ -493,25 +245,4 @@ public class BrowserLink implements Runnable } return returnValue; } - - /** - * This represents a sent message in storage. - */ - private class MessagePacket - { - public final byte[][] message; - public byte[][] returnMessage; - public Future returnValue; - - public MessagePacket(byte[][] message) - { - this(message, null); - } - - public MessagePacket(byte[][] message, Future returnValue) - { - this.message = message; - this.returnValue = returnValue; - } - } } diff --git a/addonInter/src/main/java/edu/regis/universeplayer/addon/Main.java b/addonInter/src/main/java/edu/regis/universeplayer/addon/Main.java index ae0b9ec..972fb65 100644 --- a/addonInter/src/main/java/edu/regis/universeplayer/addon/Main.java +++ b/addonInter/src/main/java/edu/regis/universeplayer/addon/Main.java @@ -7,13 +7,14 @@ package edu.regis.universeplayer.addon; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.io.IOException; import java.util.LinkedList; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; public class Main { - private static Logger logger = LoggerFactory.getLogger(Main.class); + private static final Logger logger = LoggerFactory.getLogger(Main.class); public static void main(String[] args) throws ExecutionException, InterruptedException { @@ -22,10 +23,18 @@ public class Main Thread linkThread = new Thread(link); linkThread.start(); logger.info("Sending message"); + logger.error("This is a test of the emergency logging system", new RuntimeException("This is an emergency test")); for (int i = 0, l = 20; i < l; i++) { - Future future = link.sendObject("ping"); - requests.add(future); + try + { + Future future = link.sendObject("ping"); + requests.add(future); + } + catch (IOException e) + { + logger.error("Could not send object"); + } } while (requests.size() > 0) { diff --git a/browserCommands/build.gradle b/browserCommands/build.gradle new file mode 100644 index 0000000..fa097eb --- /dev/null +++ b/browserCommands/build.gradle @@ -0,0 +1,30 @@ +/* + * Copyright (c) 2021 William Hubbard. All Rights Reserved. + */ + +plugins { + id 'java-library' +} + +// In this section you declare where to find the dependencies of your project +repositories { + mavenCentral() +} + +// In this section you declare the dependencies for your production and test code +dependencies { + // The production code uses the SLF4J logging API at compile time + implementation 'org.slf4j:slf4j-api:1.7.30' + implementation 'org.apache.logging.log4j:log4j-api:2.13.3' + implementation 'org.apache.logging.log4j:log4j-core:2.13.3' + implementation 'org.apache.logging.log4j:log4j-slf4j-impl:2.13.3' + + // Declare the dependency for your favourite test framework you want to use in your tests. + // TestNG is also supported by the Gradle Test task. Just change the + // testCompile dependency to testCompile 'org.testng:testng:6.8.1' and add + // 'test.useTestNG()' to your build script. + testImplementation 'junit:junit:4.12' +} + +sourceCompatibility = JavaVersion.VERSION_16 +targetCompatibility = JavaVersion.VERSION_16 \ No newline at end of file diff --git a/browserCommands/src/main/java/edu/regis/universeplayer/browserCommands/MessageHandler.java b/browserCommands/src/main/java/edu/regis/universeplayer/browserCommands/MessageHandler.java new file mode 100644 index 0000000..56ce186 --- /dev/null +++ b/browserCommands/src/main/java/edu/regis/universeplayer/browserCommands/MessageHandler.java @@ -0,0 +1,260 @@ +/* + * Copyright (c) 2021 William Hubbard. All Rights Reserved. + */ + +package edu.regis.universeplayer.browserCommands; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.*; +import java.nio.ByteBuffer; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedList; +import java.util.Map; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; + +/** + * + */ +public class MessageHandler implements Runnable, MessageSerializer +{ + private static final Logger logger = LoggerFactory.getLogger(MessageHandler.class); + + private final InputStream input; + private final OutputStream output; + + private final ExecutorService executor; + private final LinkedList listeners = new LinkedList<>(); + private final HashMap> messageResponses = new HashMap<>(); + + /** + * Creates a message handler. + * + * @param input - The input from our external source. + * @param output - The output to the external source. + */ + public MessageHandler(InputStream input, OutputStream output) + { + this.input = input; + this.output = output; + this.executor = Executors.newCachedThreadPool(); + } + + @Override + public void run() + { + BufferedInputStream browserIn = null; + BufferedOutputStream browserOut = null; + + boolean running = true; + + byte[][] message; + byte[] messageByte; + Object messageOb; + Future messageResponse; + ByteBuffer numBuffer = ByteBuffer.allocate(4); + HashSet toRemove = new HashSet<>(); + + try + { + browserIn = new BufferedInputStream(this.input); + browserOut = new BufferedOutputStream(this.output); + + while (running) + { + try + { + if (browserIn.available() > 0) + { + message = this.readMessage(browserIn); + if (message == null) + { + logger.info("Connection closed."); + running = false; + } + else + { + numBuffer.clear(); + numBuffer.put(message[0]); + messageOb = this.deserializeObject(message[1]); + messageResponse = this.triggerListeners(messageOb); + synchronized (this.messageResponses) + { + this.messageResponses.put(numBuffer.getInt(), messageResponse); + } + } + } + } + catch (IOException | ClassNotFoundException e) + { + logger.error("Could not retrieve message", e); + } + + /* + * Returns any finished responses. + */ + synchronized (this.messageResponses) + { + if (this.messageResponses.size() > 0) + { + for (Map.Entry> responses : this.messageResponses.entrySet()) + { + try + { + if (responses.getValue().isDone()) + { + toRemove.add(responses.getKey()); + messageByte = serializeObject(responses.getValue()); + writeMessage(browserOut, responses.getKey(), messageByte); + } + } + catch (IOException e) + { + logger.error("Could not send response message for " + responses.getKey(), e); + } + } + toRemove.forEach(this.messageResponses::remove); + toRemove.clear(); + } + } + } + } + finally + { + /* + * Release locks for any messages still waiting. + */ + synchronized (this.messageResponses) + { + this.messageResponses.values().forEach(future -> future.cancel(false)); + } + /* + * Close the streams. + */ + try + { + if (browserIn != null) + { + browserIn.close(); + } + } + catch (IOException e1) + { + logger.error("Could not close browser input", e1); + } + finally + { + try + { + if (browserOut != null) + { + browserOut.close(); + } + } + catch (IOException e1) + { + logger.error("Could not close browser output", e1); + } + } + } + } + + /** + * Triggers the message listeners. + * + * @param message - The message to send. + * @return The value returned by the listeners. + */ + protected Future triggerListeners(Object message) + { + MessageListener[] listeners; + Future returnVal; + /* + * Copies the list of listeners, so we have an unchanging array without + * holding onto the list for the whole time of execution. + */ + synchronized (this.listeners) + { + listeners = this.listeners.toArray(MessageListener[]::new); + } + returnVal = this.executor.submit(() -> { + Object returnValue = null; + for (MessageListener listener : listeners) + { + returnValue = listener.onMessage(message, returnValue); + } + return returnValue; + }); + return returnVal; + } + + /** + * Adds a message listener + * + * @param listener - A callback for when a message arrives. A value needs to + * be returned from one of the listeners. + */ + public void addListener(MessageListener listener) + { + synchronized (this.listeners) + { + this.listeners.add(listener); + } + } + + /** + * Checks to see if a listener has been added to the list. + * + * @param listener - A callback for when a message arrives. + * @return True if the provided listener is part of the list, false otherwise. + */ + public boolean hasListener(MessageListener listener) + { + synchronized (this.listeners) + { + return this.listeners.contains(listener); + } + } + + /** + * Removes a message listener from the list. + * + * @param listener - A callback for when a message arrives. + */ + public void removeListener(MessageListener listener) + { + synchronized (this.listeners) + { + this.listeners.remove(listener); + } + } + + /** + * A message listener is called when a message is sent from the remote. + * + * @see #addListener(MessageListener) + * @see #hasListener(MessageListener) + * @see #removeListener(MessageListener) + */ + public interface MessageListener + { + /** + * Called when the remote sends a message and wants a response. + * + * @param providedValue - The message sent from the remote. + * @param previousReturn The value that was returned by the previous + * listener called. A value must be returned to + * the remote. The listener either has the option + * of returning this value, or returning a new + * value. If null, then this is either the first + * listener called or none of the other listeners + * have generated return values. + * @return The value to be returned to the remote. + */ + Object onMessage(Object providedValue, Object previousReturn); + } +} diff --git a/browserCommands/src/main/java/edu/regis/universeplayer/browserCommands/MessageRunner.java b/browserCommands/src/main/java/edu/regis/universeplayer/browserCommands/MessageRunner.java new file mode 100644 index 0000000..ea5b6a8 --- /dev/null +++ b/browserCommands/src/main/java/edu/regis/universeplayer/browserCommands/MessageRunner.java @@ -0,0 +1,372 @@ +/* + * Copyright (c) 2021 William Hubbard. All Rights Reserved. + */ + +package edu.regis.universeplayer.browserCommands; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.*; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.util.HashMap; +import java.util.LinkedList; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +/** + * This class serves as a general inter-process object messaging system. + */ +public abstract class MessageRunner implements Runnable, MessageSerializer +{ + private static final Logger logger = LoggerFactory.getLogger(MessageRunner.class); + + private final InputStream input; + private final OutputStream output; + private final Object readLock = new Object(); + + /** + * This queue serves as a cache for objects we need to send. + */ + private final LinkedList sendQueue = new LinkedList<>(); + /** + * This queue serves as a cache for objects that are waiting for a response. + */ + private final HashMap sentQueue = new HashMap<>(); + private int messagesSent = 0; + + /** + * Creates a message runner. + * + * @param input - The input from our external source. + * @param output - The output to the external source. + */ + public MessageRunner(InputStream input, OutputStream output) + { + this.input = input; + this.output = output; + } + + @Override + public void run() + { + BufferedInputStream browserIn = null; + BufferedOutputStream browserOut = null; + MessagePacket packet; + + boolean running = true; + + byte[][] returnMessage; + ByteBuffer numBuffer = ByteBuffer.allocate(4); + + try + { + browserIn = new BufferedInputStream(this.input); + browserOut = new BufferedOutputStream(this.output); + + while (running) + { + /* + * Sends a messages + */ + synchronized (this.sendQueue) + { + packet = this.sendQueue.poll(); + } + if (packet != null) + { + try + { + packet.returnValue.index = this.messagesSent; + writeMessage(browserOut, this.messagesSent, packet.message); + + synchronized (this.sentQueue) + { + this.sentQueue.put(this.messagesSent, packet); + this.messagesSent++; + } + } + catch (IOException e) + { + logger.error("Could not send message " + new String(packet.message, StandardCharsets.UTF_8), e); + } + } + + /* + * Reads any available messages + */ + try + { + if (!this.sentQueue.isEmpty()) + { + if (browserIn.available() > 0) + { + /* + * Wait for a response from the browser. + */ + returnMessage = this.readMessage(browserIn); + if (returnMessage == null) + { + logger.info("Connection closed."); + running = false; + } + else + { + numBuffer.clear(); + numBuffer.put(returnMessage[0]); + numBuffer.clear(); + packet = this.sentQueue.get(numBuffer.getInt()); + if (packet == null) + { + numBuffer.clear(); + logger.warn("Received message {} for nonexistant packet", numBuffer.getInt()); + } + else + { + packet.returnMessage = returnMessage[1]; + numBuffer.clear(); + logger.debug("Reading message {} {}", numBuffer.getInt(), packet.returnMessage); + synchronized (this.readLock) + { + logger.trace("Received message {}, notifying futures.", packet.returnValue.index); + this.readLock.notifyAll(); + } + } + } + } + } + } + catch (IOException e) + { + logger.error("Could not retrieve message", e); + } + } + } + finally + { + /* + * Release locks for any messages still waiting. + */ + synchronized (this.sendQueue) + { + while (this.sendQueue.size() > 0) + { + packet = this.sendQueue.poll(); + packet.returnMessage = null; + } + } + synchronized (this.readLock) + { + this.readLock.notifyAll(); + } + /* + * Close the streams. + */ + try + { + if (browserIn != null) + { + browserIn.close(); + } + } + catch (IOException e1) + { + logger.error("Could not close browser input", e1); + } + finally + { + try + { + if (browserOut != null) + { + browserOut.close(); + } + } + catch (IOException e1) + { + logger.error("Could not close browser output", e1); + } + } + } + } + + /** + * Sends an object to the remote. + * + * @param message - The message to send. + * @return A future for whatever the remote may return. + * @throws IOException Should an error occur in serializing the object. + */ + public Future sendObject(Object message) throws IOException + { + byte[] messageData = this.serializeObject(message); + MessagePacket packetEntry = new MessagePacket(messageData); + MessageFuture future = new MessageFuture(packetEntry); + + synchronized (this.sendQueue) + { + sendQueue.add(packetEntry); + } + + return future; + } + + /** + * This represents a sent message in storage. + */ + public static class MessagePacket + { + public final byte[] message; + public byte[] returnMessage; + public MessageFuture returnValue; + + public MessagePacket(byte[] message) + { + this(message, null); + } + + public MessagePacket(byte[] message, MessageFuture returnValue) + { + this.message = message; + this.returnValue = returnValue; + } + } + + /** + * This future waits for a return message from the runner. + */ + public class MessageFuture implements Future + { + private final MessagePacket packetEntry; + private int index; + private boolean canceled = false; + + private MessageFuture(MessagePacket packet) + { + this.packetEntry = packet; + packet.returnValue = this; + } + + @Override + public boolean cancel(boolean mayInterruptIfRunning) + { + synchronized (sendQueue) + { + if (sendQueue.remove(this.packetEntry)) + { + this.canceled = true; + if (mayInterruptIfRunning) + { + logger.trace("Canceling future, notifying self futures."); + synchronized (sentQueue) + { + sentQueue.remove(this.index); + } + synchronized (readLock) + { + readLock.notifyAll(); + } + } + else + { + logger.trace("Canceling future."); + } + return true; + } + else + { + return false; + } + } + } + + @Override + public boolean isCancelled() + { + return this.canceled; + } + + @Override + public boolean isDone() + { + return this.packetEntry.returnMessage != null; + } + + @Override + public Object get() throws InterruptedException, ExecutionException + { + if (this.isDone()) + { + try + { + return deserializeObject(this.packetEntry.returnMessage); + } + catch (IOException | ClassNotFoundException e) + { + throw new ExecutionException("Error parsing object", e); + } + } + synchronized (readLock) + { + while (!this.canceled && packetEntry.returnMessage == null) + { + logger.trace("Future {} waiting for message", this.index); + readLock.wait(); + } + } + if (packetEntry.returnMessage == null) + { + throw new InterruptedException(); + } + logger.trace("Future {} has received message", this.index); + try + { + return deserializeObject(this.packetEntry.returnMessage); + } + catch (IOException | ClassNotFoundException e) + { + throw new ExecutionException("Error parsing object", e); + } + } + + @Override + public Object get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException + { + if (this.isDone()) + { + try + { + return deserializeObject(this.packetEntry.returnMessage); + } + catch (IOException | ClassNotFoundException e) + { + throw new ExecutionException("Error parsing object", e); + } + } + synchronized (readLock) + { + while (!this.canceled && packetEntry.returnMessage == null) + { + logger.trace("Future {} waiting for message", this.index); + readLock.wait(unit.toMillis(timeout)); + } + } + if (packetEntry.returnMessage == null) + { + throw new InterruptedException(); + } + logger.trace("Future {} has received message", this.index); + try + { + return deserializeObject(this.packetEntry.returnMessage); + } + catch (IOException | ClassNotFoundException e) + { + throw new ExecutionException("Error parsing object", e); + } + } + } +} diff --git a/browserCommands/src/main/java/edu/regis/universeplayer/browserCommands/MessageSerializer.java b/browserCommands/src/main/java/edu/regis/universeplayer/browserCommands/MessageSerializer.java new file mode 100644 index 0000000..0fd5730 --- /dev/null +++ b/browserCommands/src/main/java/edu/regis/universeplayer/browserCommands/MessageSerializer.java @@ -0,0 +1,138 @@ +/* + * Copyright (c) 2021 William Hubbard. All Rights Reserved. + */ + +package edu.regis.universeplayer.browserCommands; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.*; +import java.nio.ByteBuffer; + +public interface MessageSerializer +{ + Logger logger = LoggerFactory.getLogger(MessageSerializer.class); + + /** + * Converts an object into a form that can be sent. + * + * @param message - The message to send. + * @return The byte array representation of that object. + * @throws IOException Should a serialization error occur. + */ + default byte[] serializeObject(Object message) throws IOException + { + try (ByteArrayOutputStream byteStream = new ByteArrayOutputStream()) + { + try (ObjectOutputStream stream = new ObjectOutputStream(byteStream)) + { + stream.writeObject(message); + } + return byteStream.toByteArray(); + } + } + + /** + * Converts a byte stream into an object. + * + * @param message - The message received. + * @return An object. + * @throws IOException If there is an error in parsing the message. + * @throws ClassNotFoundException If the object is not recognized + */ + default Object deserializeObject(byte[] message) throws IOException, ClassNotFoundException + { + try (ByteArrayInputStream byteStream = new ByteArrayInputStream(message)) + { + try (ObjectInputStream stream = new ObjectInputStream(byteStream)) + { + return stream.readObject(); + } + } + } + + /** + * Writes a message to the output stream. + * + * @param out - The output stream to write to. + * @param messageNum - The ID of the message being sent. This will help keep track of responses. + * @param message - The actual message contents to write. + * @throws IOException Thrown when an exception occures + */ + default void writeMessage(OutputStream out, int messageNum, byte[] message) throws IOException + { + ByteBuffer lengthBuffer = ByteBuffer.allocate(4); + logger.debug("Writing message {} {}", messageNum, message); + /* + * Writes the message number. + */ + lengthBuffer.clear(); + lengthBuffer.putInt(messageNum); + out.write(lengthBuffer.array()); + /* + * Writes the message length + */ + /* + * Wipe out the buffer + */ + lengthBuffer.clear(); + lengthBuffer.put(new byte[4]); + lengthBuffer.clear(); + lengthBuffer.putInt(message.length); + out.write(lengthBuffer.array()); + /* + * Writes the message + */ + out.write(message); + out.flush(); + } + + /** + * Reads a message from the input stream + * + * @param in - The input stream to read from. + * @return Two byte arrays, each with their own value encoded. The first is + * a 4-byte representation of the message number this response corresponds + * to. The second array is the message itself. If the stream has been + * closed, null is returned. + * @throws IOException Thrown when an exception occurs reading a message. + */ + default byte[][] readMessage(InputStream in) throws IOException + { + int readLength; + ByteBuffer lengthBuffer = ByteBuffer.allocate(4); + byte[] messageNum = new byte[4]; + byte[] message; + /* + * Reads the message number. + */ + readLength = in.read(messageNum); + if (readLength == 0) + { + logger.debug("Input stream closed, no more messages."); + return null; + } + /* + * Reads the message length + */ + lengthBuffer.clear(); + readLength = in.read(lengthBuffer.array()); + if (readLength == 0) + { + logger.error("Malformed message, could not get message length."); + return null; + } + message = new byte[lengthBuffer.getInt()]; + /* + * Writes the message + */ + readLength = in.read(message); + if (readLength < message.length) + { + logger.warn("Message shorter than reported (expected {} bytes, got {} bytes)", message.length, readLength); + } + logger.debug("Reading message"); + return new byte[][] {messageNum, message}; + } +} diff --git a/interface/build.gradle b/interface/build.gradle index 1c22872..a2e9894 100644 --- a/interface/build.gradle +++ b/interface/build.gradle @@ -24,6 +24,7 @@ dependencies { implementation 'com.googlecode.soundlibs:jlayer:1.0.1.4' implementation 'com.mpatric:mp3agic:0.9.1' implementation project(":libwave") + implementation project(":browserCommands") // Declare the dependency for your favourite test framework you want to use in your tests. // TestNG is also supported by the Gradle Test task. Just change the diff --git a/settings.gradle b/settings.gradle index 077a544..bbbfa0b 100644 --- a/settings.gradle +++ b/settings.gradle @@ -23,5 +23,6 @@ rootProject.name = 'UniversalMusicPlayer' include ':interface' include ':add-on' include ':addonInter' +include ':browserCommands' include ':player' include ':libwave'