import java.io.*;
import java.net.*;
public class Node {
public static void main(String[] args) throws IOException{
//Create first instance of server class by declaring two arguments
if(args.length == 2) {
Server s = new Server(Integer.parseInt(args[0]),Integer.parseInt(args[1]));
s.startListening();
}
//Next while server is listening create first instance of client class by declaring four arguments
else if(args.length == 4) {
Client c = new Client(Integer.parseInt(args[0]),args[1],Integer.parseInt(args[2]),args[3]);
c.connect();
}
//if no arguments are declared display error messages
else {
System.err.println("\nUsage (Server): java Node <port> <clients>");
System.err.println("Usage (Client): java Node <id> <server> <port> <COMMIT|ABORT>\n");
System.err.println("Note: Only ports 9000-9060 may be used on DCS machines\n");
System.exit(-1);
}
}
}
class Server {
//server object private variable
private int port;
private int clients;
private ServerSocket serverSocket = null;
private int client_connections = 0; // number of clients connected at time t
private int aborts = 0;
private int commits = 0;
private String message;
connectionThread[] clientConnections; // array to keep track of clients
//Server object construct
Server(int port, int clients){
this.port = port;
this.clients = clients;
clientConnections = new connectionThread[clients];
}
public void startListening() throws IOException{
try {
//Create new socket object to listen to the specific given port
serverSocket = new ServerSocket(port);
System.out.println("<Server>: I'm alive and waiting for " + clients + " clients");
}
catch (IOException e) {
System.out.println("Read failed");
//e.printStackTrace();
}
while(!(client_connections == clients)) {
clientConnections[client_connections] = new connectionThread(serverSocket.accept(), this);
clientConnections[client_connections].start();
client_inc(); }
System.out.println("All clients have connected");
// all clients have now been connected
// time to propose a commit
stop_listening();
broadcast("server proposes commit");
while ( (commits+aborts) != clients) { for (int i=0;i<100000;i++){} } // wait until all clients have responded
if (aborts > 0 )
broadcast("server decides abort");
else broadcast("server decides commit");
}
private void broadcast(String message) { for (int i=0;i<clients;i++) clientConnections[i].outputMessage(message);}
public void client_inc () { client_connections++; }
private void stop_listening () { try {serverSocket.close();} catch (IOException e) {System.out.println("couldnt close server"); System.exit(-1); } }
public void aborts_inc() {aborts++;}
public void commits_inc() {commits++;}
}
class connectionThread extends Thread{
private Socket clientSocket = null;
private String message;
Server parent_server;
ObjectOutputStream out;
ObjectInputStream in;
public connectionThread(Socket clientSocket, Server server){
this.clientSocket = clientSocket;
parent_server = server;
}
public void run(){
try {
out = new ObjectOutputStream(clientSocket.getOutputStream());
out.flush(); //write any buffered output and flushes to underlying stream
in = new ObjectInputStream(clientSocket.getInputStream());
}
catch (IOException e) {
System.out.println("Read failed");
//e.printStackTrace();
}
do{
try{
//communication to/from client
message = (String)in.readObject();
System.out.println("<Client>: " + message);
outputMessage("Client connection accepted");
message = (String)in.readObject();
System.out.println("<Client>: " + message);
if (message.equals("Sending [Hello] message to client " + extractId(message))){
outputMessage("[Hello]");
outputMessage("Sending [Ack] message to client " + extractId(message));
}
message = (String)in.readObject();
System.out.println("<Client>: " + message);
//End communication
message = "End";
if (message.equals("End"))
break;
}
catch(ClassNotFoundException classnot){
System.err.println("Data received in unknown format");
}
catch (IOException e) {
e.printStackTrace();
}
}while(!message.equals("End"));
// my code for the protocol goes here
//parent_server.client_inc();
while(true) {
try {message = (String)in.readObject();} catch (IOException e) {System.exit(-1);}
catch (ClassNotFoundException classnot) {System.exit(-1);}
if (message.equals("i have status commit"))
{parent_server.commits_inc(); break; }
else if (message.equals("i have status abort"))
{parent_server.aborts_inc(); break;}
}
//Close all socket, input and output stream connections
try{
in.close();
out.close();
clientSocket.close();
}
catch(IOException ioException){
ioException.printStackTrace();
}
}
//method to output messages to client
public void outputMessage(String msg) {
try{
out.writeObject(msg);
out.flush();
System.out.println("<Server>: " + msg);
} catch(IOException ioException){ ioException.printStackTrace(); }
}
public String extractId(String message){
String words = message;
String findId = "";
int track = 0;
for(int i = 0; i < words.length() - 1 ; i++){
if (words.charAt(i) == ' ')track = i;
}
for (int i = track + 1; i < words.length(); i++){
findId = findId + words.charAt(i);
}
return findId;
}
}
class Client {
//Client object variables
private int id;
private String host;
private int port;
private String status;
String message;
boolean listening = true;
Socket Soc = null;
ObjectOutputStream out = null;
ObjectInputStream in = null;
//Client object construct
Client(int id, String host, int port, String status) {
this.id = id;
this.host = host;
this.port = port;
this.status = status;
}
public void connect() throws IOException{
try{
//create new socket
Soc = new Socket(host, port);
//create input and output streams
out = new ObjectOutputStream(Soc.getOutputStream());
out.flush();
in = new ObjectInputStream(Soc.getInputStream());
}
catch (UnknownHostException e) {
System.err.println("Host is unreachable!");
System.exit(1);
} catch (IOException e) {
System.out.println("No I/O");
//e.printStackTrace();
System.exit(1);
}
/**
* In order to initial communication one class must first send a message before their while loops and
* in order for the other class to receive the message they must declare their first message in their
* while loops with the in.readObject();, In this case the client class has initiated communications
* with the server.
*/
//initiate communication
outputMessage("I'm Alive and trying to connect to server");
do{
try{
//Communication with server
message = (String)in.readObject();
System.out.println("<Server>: " + message);
outputMessage("Sending [Hello] message to client " + id);
message = (String)in.readObject();
System.out.println("<Server>: " + message);
message = (String)in.readObject();
System.out.println("<Server>: " + message);
if (message.equals("Sending [Ack] message to client " + id))
outputMessage("[Ack]");
System.out.println();
//End communication
message = "End";
outputMessage(message);
}
catch(ClassNotFoundException classNot){
System.err.println("data received in unknown format");
}
}while(!message.equals("End"));
while(true) {
try {message = (String)in.readObject();} catch (IOException e) {System.exit(-1);}
catch (ClassNotFoundException classnot) {System.exit(-1);}
if (message.equals("server proposes abort")) {
if (status.equals("COMMIT"))
outputMessage("i have status commit");
else if (status.equals("ABORT"))
outputMessage("i have status abort"); }
if (message.equals("server decides commit"))
{ System.out.println("<Client " + id + " > Server said we were commiting"); break; }
else if (message.equals("server decides abort"))
{ System.out.println("<Client " + id + " > Server said we were aborting"); break; }
// listen for "server proposes abort //i have status commit or i have status abort
// server decides commit
// server decides abort
}
//Close all socket, input and output stream connections
try{
in.close();
out.close();
Soc.close();
}
catch(IOException ioException){
ioException.printStackTrace();
}
}
//method to output messages to server
void outputMessage(String msg)throws IOException
{
try{
out.writeObject(msg);
out.flush();
System.out.println("<Client " + id+ " >: " + msg);
}
catch(IOException e){
e.printStackTrace();
}
}
public int getClientId(){
return id;
}
}