Implemented MessageRouter.
Added ConnectionToServerCommandFactory, and JoinSuccessfulCommand, RaceParticipantsFullCOmmand, ServerFullCommand. Added IncomingHeartBeatCommandFactory, and IncomingHeartBeatCommand. Added ConnectionToServerState, which represents the client's connection state to server. Renamed VisualiserInput to VisualiserRaceController. Added ConnectionToServer, which tracks the client's connection to server. Added ConnectionToServerController, which accepts JoinAcceptance messages, turns them into commands, and passes them to ConnectionToServer. Added IncomingHeartBeatService, which tracks the heart beat status of the connection. Added IncomingHeartBeatController, which accepts HeartBeat messages, turns them into commands, and passes them to IncomingHeartBeatService. Refactored ServerConnection a bit. #story[1095]
This commit is contained in:
fjc40
@@ -1,243 +0,0 @@
|
||||
package visualiser.model;
|
||||
|
||||
|
||||
import mock.enums.ConnectionStateEnum;
|
||||
import network.Messages.AC35Data;
|
||||
import network.Messages.Enums.JoinAcceptanceEnum;
|
||||
import network.Messages.Enums.MessageType;
|
||||
import network.Messages.Enums.RequestToJoinEnum;
|
||||
import network.Messages.JoinAcceptance;
|
||||
import network.Messages.LatestMessages;
|
||||
import network.Messages.RequestToJoin;
|
||||
import network.StreamRelated.MessageDeserialiser;
|
||||
import network.StreamRelated.MessageSerialiser;
|
||||
import shared.exceptions.HandshakeException;
|
||||
import visualiser.app.VisualiserInput;
|
||||
import visualiser.gameController.ControllerClient;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.Socket;
|
||||
import java.util.concurrent.BlockingQueue;
|
||||
import java.util.concurrent.LinkedBlockingQueue;
|
||||
import java.util.logging.Level;
|
||||
import java.util.logging.Logger;
|
||||
|
||||
/**
|
||||
* This class handles the client-server connection handshake, and creation of VisualiserInput and ControllerClient.
|
||||
*/
|
||||
public class ServerConnection implements Runnable {
|
||||
|
||||
/**
|
||||
* The socket for the connection to server.
|
||||
*/
|
||||
private Socket socket;
|
||||
|
||||
/**
|
||||
* The source ID that has been allocated to the client.
|
||||
*/
|
||||
private int allocatedSourceID = 0;
|
||||
|
||||
|
||||
/**
|
||||
* Latest snapshot of the race, received from the server.
|
||||
*/
|
||||
private LatestMessages latestMessages;
|
||||
|
||||
|
||||
/**
|
||||
* Used to convert incoming messages into a race snapshot.
|
||||
*/
|
||||
private VisualiserInput visualiserInput;
|
||||
|
||||
/**
|
||||
* Used to send client input to server.
|
||||
*/
|
||||
private ControllerClient controllerClient;
|
||||
|
||||
|
||||
/**
|
||||
* Used to write messages to socket.
|
||||
*/
|
||||
private MessageSerialiser messageSerialiser;
|
||||
|
||||
/**
|
||||
* Stores messages to write to socket.
|
||||
*/
|
||||
private BlockingQueue<AC35Data> outputQueue;
|
||||
|
||||
/**
|
||||
* Used to read messages from socket.
|
||||
*/
|
||||
private MessageDeserialiser messageDeserialiser;
|
||||
|
||||
/**
|
||||
* Stores messages read from socket.
|
||||
*/
|
||||
private BlockingQueue<AC35Data> inputQueue;
|
||||
|
||||
/**
|
||||
* The state of the connection to the client.
|
||||
*/
|
||||
private ConnectionStateEnum connectionState = ConnectionStateEnum.UNKNOWN;
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
/**
|
||||
* Creates a server connection, using a given socket.
|
||||
* @param socket The socket which connects to the client.
|
||||
* @param latestMessages Latest race snapshot to send to client.
|
||||
* @throws IOException Thrown if there is a problem with the client socket.
|
||||
*/
|
||||
public ServerConnection(Socket socket, LatestMessages latestMessages) throws IOException {
|
||||
this.socket = socket;
|
||||
this.latestMessages = latestMessages;
|
||||
|
||||
this.outputQueue = new LinkedBlockingQueue<>();
|
||||
this.inputQueue = new LinkedBlockingQueue<>();
|
||||
|
||||
|
||||
this.messageSerialiser = new MessageSerialiser(socket.getOutputStream(), outputQueue);
|
||||
this.messageDeserialiser = new MessageDeserialiser(socket.getInputStream(), inputQueue);
|
||||
|
||||
new Thread(messageSerialiser, "ServerConnection()->MessageSerialiser thread " + messageSerialiser).start();
|
||||
new Thread(messageDeserialiser, "ServerConnection()->MessageDeserialiser thread " + messageDeserialiser).start();
|
||||
|
||||
|
||||
this.controllerClient = new ControllerClient(outputQueue);
|
||||
|
||||
}
|
||||
|
||||
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
try {
|
||||
handshake();
|
||||
|
||||
} catch (HandshakeException e) {
|
||||
Logger.getGlobal().log(Level.WARNING, "Server handshake failed.", e);
|
||||
Thread.currentThread().interrupt();
|
||||
return;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Initiates the handshake with the server.
|
||||
* @throws HandshakeException Thrown if something goes wrong with the handshake.
|
||||
*/
|
||||
private void handshake() throws HandshakeException {
|
||||
|
||||
//This function is a bit messy, and could probably be refactored a bit.
|
||||
|
||||
connectionState = ConnectionStateEnum.WAITING_FOR_HANDSHAKE;
|
||||
|
||||
|
||||
sendRequestToJoinMessage(RequestToJoinEnum.PARTICIPANT);
|
||||
|
||||
|
||||
JoinAcceptance joinAcceptance = waitForJoinAcceptance();
|
||||
|
||||
|
||||
//If we join successfully...
|
||||
if (joinAcceptance.getAcceptanceType() == JoinAcceptanceEnum.JOIN_SUCCESSFUL) {
|
||||
|
||||
this.allocatedSourceID = joinAcceptance.getSourceID();
|
||||
|
||||
|
||||
//new Thread(controllerClient, "ServerConnection.run()->ControllerClient thread " + controllerClient).start();
|
||||
|
||||
}
|
||||
|
||||
this.visualiserInput = new VisualiserInput(latestMessages, inputQueue);
|
||||
new Thread(visualiserInput, "ServerConnection.run()->VisualiserInput thread " + visualiserInput).start();
|
||||
|
||||
|
||||
connectionState = ConnectionStateEnum.CONNECTED;
|
||||
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Waits until the server sends a {@link JoinAcceptance} message, and returns it.
|
||||
* @return The {@link JoinAcceptance} message.
|
||||
* @throws HandshakeException Thrown if we get interrupted while waiting.
|
||||
*/
|
||||
private JoinAcceptance waitForJoinAcceptance() throws HandshakeException {
|
||||
|
||||
try {
|
||||
|
||||
|
||||
while (connectionState == ConnectionStateEnum.WAITING_FOR_HANDSHAKE) {
|
||||
|
||||
AC35Data message = inputQueue.take();
|
||||
|
||||
//We need to wait until they actually send a join request.
|
||||
if (message.getType() == MessageType.JOIN_ACCEPTANCE) {
|
||||
return (JoinAcceptance) message;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
throw new HandshakeException("Handshake failed. Thread: " + Thread.currentThread() + " was interrupted while waiting on the incoming message queue.", e);
|
||||
|
||||
}
|
||||
|
||||
|
||||
throw new HandshakeException("Handshake was cancelled. Connection state is now: " + connectionState);
|
||||
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Sends the server a {@link RequestToJoin} message.
|
||||
* @param requestType The type of request to send
|
||||
* @throws HandshakeException Thrown if the thread is interrupted while placing message on the outgoing message queue.
|
||||
*/
|
||||
private void sendRequestToJoinMessage(RequestToJoinEnum requestType) throws HandshakeException {
|
||||
|
||||
//Send them the source ID.
|
||||
RequestToJoin requestToJoin = new RequestToJoin(requestType);
|
||||
|
||||
try {
|
||||
outputQueue.put(requestToJoin);
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
throw new HandshakeException("Handshake failed. Thread: " + Thread.currentThread() + " interrupted while placing RequestToJoin message on outgoing message queue.", e);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Determines whether or not this connection is still alive.
|
||||
* This is based off whether the {@link MessageDeserialiser} is still alive.
|
||||
* @return True if it is alive, false otherwise.
|
||||
*/
|
||||
public boolean isAlive() {
|
||||
return messageDeserialiser.isRunning();
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Returns the controller client, which writes BoatAction messages to the outgoing queue.
|
||||
* @return The ControllerClient.
|
||||
*/
|
||||
public ControllerClient getControllerClient() {
|
||||
return controllerClient;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the source ID that has been allocated to the client.
|
||||
* @return Source ID allocated to the client. 0 if it hasn't been allocated.
|
||||
*/
|
||||
public int getAllocatedSourceID() {
|
||||
return allocatedSourceID;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,213 @@
|
||||
package visualiser.model;
|
||||
import network.Messages.*;
|
||||
import shared.model.RunnableWithFramePeriod;
|
||||
|
||||
import java.util.concurrent.BlockingQueue;
|
||||
import java.util.logging.Level;
|
||||
import java.util.logging.Logger;
|
||||
|
||||
|
||||
/**
|
||||
* TCP client which receives packets/messages from a race data source
|
||||
* (e.g., mock source, official source), and exposes them to any observers.
|
||||
*/
|
||||
public class VisualiserRaceController implements RunnableWithFramePeriod {
|
||||
|
||||
|
||||
/**
|
||||
* Incoming messages from server.
|
||||
*/
|
||||
private BlockingQueue<AC35Data> incomingMessages;
|
||||
|
||||
|
||||
/**
|
||||
* An object containing the set of latest messages to write to.
|
||||
* Every server frame, VisualiserInput reads messages from its incomingMessages, and write them to this.
|
||||
*/
|
||||
private LatestMessages latestMessages;
|
||||
|
||||
|
||||
/**
|
||||
* Constructs a visualiserInput to convert an incoming stream of messages into LatestMessages.
|
||||
*
|
||||
* @param latestMessages Object to place messages in.
|
||||
* @param incomingMessages The incoming queue of messages.
|
||||
*/
|
||||
public VisualiserRaceController(LatestMessages latestMessages, BlockingQueue<AC35Data> incomingMessages) {
|
||||
this.latestMessages = latestMessages;
|
||||
this.incomingMessages = incomingMessages;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Returns the LatestMessages object, which can be queried for any received race related messages.
|
||||
*
|
||||
* @return The LatestMessages object.
|
||||
*/
|
||||
public LatestMessages getLatestMessages() {
|
||||
return latestMessages;
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
|
||||
while (!Thread.interrupted()) {
|
||||
|
||||
AC35Data message = null;
|
||||
try {
|
||||
message = incomingMessages.take();
|
||||
} catch (InterruptedException e) {
|
||||
Logger.getGlobal().log(Level.SEVERE, "VisualiserInput was interrupted on thread: " + Thread.currentThread() + " while waiting for messages.");
|
||||
Thread.currentThread().interrupt();
|
||||
return;
|
||||
}
|
||||
|
||||
|
||||
//TODO refactor below
|
||||
|
||||
//Checks which message is being received and does what is needed for that message.
|
||||
switch (message.getType()) {
|
||||
|
||||
|
||||
//RaceStatus.
|
||||
case RACESTATUS: {
|
||||
RaceStatus raceStatus = (RaceStatus) message;
|
||||
|
||||
//System.out.println("Race Status Message");
|
||||
this.latestMessages.setRaceStatus(raceStatus);
|
||||
|
||||
for (BoatStatus boatStatus : raceStatus.getBoatStatuses()) {
|
||||
this.latestMessages.setBoatStatus(boatStatus);
|
||||
}
|
||||
|
||||
break;
|
||||
}
|
||||
|
||||
//DisplayTextMessage.
|
||||
case DISPLAYTEXTMESSAGE: {
|
||||
//System.out.println("Display Text Message");
|
||||
//No decoder for this.
|
||||
|
||||
break;
|
||||
}
|
||||
|
||||
//XMLMessage.
|
||||
case XMLMESSAGE: {
|
||||
XMLMessage xmlMessage = (XMLMessage) message;
|
||||
|
||||
//System.out.println("XML Message!");
|
||||
|
||||
this.latestMessages.setXMLMessage(xmlMessage);
|
||||
|
||||
break;
|
||||
}
|
||||
|
||||
//RaceStartStatus.
|
||||
case RACESTARTSTATUS: {
|
||||
|
||||
//System.out.println("Race Start Status Message");
|
||||
|
||||
break;
|
||||
}
|
||||
|
||||
//YachtEventCode.
|
||||
case YACHTEVENTCODE: {
|
||||
//YachtEventCode yachtEventCode = (YachtEventCode) message;
|
||||
|
||||
//System.out.println("Yacht Event Code!");
|
||||
//No decoder for this.
|
||||
|
||||
break;
|
||||
}
|
||||
|
||||
//YachtActionCode.
|
||||
case YACHTACTIONCODE: {
|
||||
//YachtActionCode yachtActionCode = (YachtActionCode) message;
|
||||
|
||||
//System.out.println("Yacht Action Code!");
|
||||
// No decoder for this.
|
||||
|
||||
break;
|
||||
}
|
||||
|
||||
//ChatterText.
|
||||
case CHATTERTEXT: {
|
||||
//ChatterText chatterText = (ChatterText) message;
|
||||
|
||||
//System.out.println("Chatter Text Message!");
|
||||
//No decoder for this.
|
||||
|
||||
break;
|
||||
}
|
||||
|
||||
//BoatLocation.
|
||||
case BOATLOCATION: {
|
||||
BoatLocation boatLocation = (BoatLocation) message;
|
||||
|
||||
//System.out.println("Boat Location!");
|
||||
|
||||
BoatLocation existingBoatLocation = this.latestMessages.getBoatLocationMap().get(boatLocation.getSourceID());
|
||||
if (existingBoatLocation != null) {
|
||||
//If our boatlocation map already contains a boat location message for this boat, check that the new message is actually for a later timestamp (i.e., newer).
|
||||
if (boatLocation.getTime() > existingBoatLocation.getTime()) {
|
||||
//If it is, replace the old message.
|
||||
this.latestMessages.setBoatLocation(boatLocation);
|
||||
}
|
||||
} else {
|
||||
//If the map _doesn't_ already contain a message for this boat, insert the message.
|
||||
this.latestMessages.setBoatLocation(boatLocation);
|
||||
}
|
||||
|
||||
break;
|
||||
}
|
||||
|
||||
//MarkRounding.
|
||||
case MARKROUNDING: {
|
||||
MarkRounding markRounding = (MarkRounding) message;
|
||||
|
||||
//System.out.println("Mark Rounding Message!");
|
||||
|
||||
MarkRounding existingMarkRounding = this.latestMessages.getMarkRoundingMap().get(markRounding.getSourceID());
|
||||
if (existingMarkRounding != null) {
|
||||
|
||||
//If our markRoundingMap already contains a mark rounding message for this boat, check that the new message is actually for a later timestamp (i.e., newer).
|
||||
if (markRounding.getTime() > existingMarkRounding.getTime()) {
|
||||
//If it is, replace the old message.
|
||||
this.latestMessages.setMarkRounding(markRounding);
|
||||
}
|
||||
|
||||
} else {
|
||||
//If the map _doesn't_ already contain a message for this boat, insert the message.
|
||||
this.latestMessages.setMarkRounding(markRounding);
|
||||
}
|
||||
|
||||
break;
|
||||
}
|
||||
|
||||
//CourseWinds.
|
||||
case COURSEWIND: {
|
||||
|
||||
//System.out.println("Course Wind Message!");
|
||||
CourseWinds courseWinds = (CourseWinds) message;
|
||||
|
||||
this.latestMessages.setCourseWinds(courseWinds);
|
||||
|
||||
break;
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
||||
//Main loop.
|
||||
// take message
|
||||
// create command
|
||||
// place in command queue
|
||||
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user