Skip to content
Snippets Groups Projects
Commit c7292a34 authored by tmp1u19's avatar tmp1u19 :octopus:
Browse files

Add storing operation in Dstore

parent de9967ca
Branches
No related tags found
No related merge requests found
Hide whitespace changes
Inline Side-by-side
src/Controller.java
+ 63
− 25
View file @ c7292a34
import java.io.*;
import java.net.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
public class Controller {
......@@ -10,7 +11,7 @@ public class Controller {
// waits for Dstores to join the datastore (rebalance operation)
// it doesn't serve any client request until at least R Dstores have joined the system
File file = new File("output.txt");
static File file = new File("output.txt");
public static void main (String[] args) {
......@@ -24,7 +25,15 @@ public class Controller {
int rebalance_period = Integer.parseInt(args[3]);
System.out.println("Started");
// list of all dstores
List<Socket> ports = new ArrayList<>();
// file, list of dstores
HashMap<String, List<Socket>> index = new HashMap<>();
// file, status (e.g. in progress)
HashMap<String, String> status = new HashMap<>();
createLogFile();
try {
ServerSocket socket = new ServerSocket(cport);
......@@ -37,31 +46,32 @@ public class Controller {
ports.add(client);
R = R - 1;
Thread t = new Thread() {
@Override
public void run() {
try {
handleDstores(client);
} catch (Exception e) {
e.printStackTrace();
}
}
};
t.start();
} else {
//Socket clientDstore = socket.accept();
for(Socket client : ports) {
Thread t = new Thread() {
@Override
public void run() {
try {
handleDstores(client);
} catch (Exception e) {
e.printStackTrace();
}
}
};
t.start();
}
break;
}
} catch (Exception e1) {
System.out.println(e1);
}
}
/*
#TODO
- use threads to connect multiple clients (Dstore and Client) to a server
*/
} catch (IOException e) {
System.out.println("error" + e);
......@@ -81,6 +91,7 @@ public class Controller {
if(cmd.equals("ACK")) {
} else if (cmd.equals("STORE_ACK")) {
System.out.println("STORE_ACK");
} else if (cmd.equals("REMOVE_ACK")) {
......@@ -88,12 +99,13 @@ public class Controller {
clientSocket.close();
break;
} else {
String msg = "unknown command " + line + "\n";
outputStream.write(msg.getBytes());
String msg = "unknown command " + line;
// outputStream.write(msg.getBytes());
log(msg);
}
String msg = "You typed " + line + "\n";
outputStream.write(msg.getBytes());
String msg = "You typed " + line;
log(msg);
}
if(!clientSocket.isClosed()) {
......@@ -101,7 +113,32 @@ public class Controller {
}
}
private void createLogFile() {
private static void handleClient(Socket clientSocket) throws IOException {
OutputStream outputStream = clientSocket.getOutputStream();
InputStream inputStream = clientSocket.getInputStream();
BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));
String line;
while((line = reader.readLine()) != null) {
String[] tokens = line.split(" ");
String cmd = tokens[0];
if (cmd.equals("STORE")) {
} else if (cmd.equals("LOAD")) {
} else if (cmd.equals("REMOVE")) {
} else if (cmd.equals("LIST")) {
} else {
String msg = "unknown command " + line;
log(msg);
}
}
}
private static void createLogFile() {
try {
if(file.createNewFile()) {
System.out.println("File created");
......@@ -113,10 +150,11 @@ public class Controller {
}
}
private void log(String message) {
private static void log(String message) {
try {
FileWriter writer = new FileWriter(file);
writer.write(message + " at " + System.currentTimeMillis() + "\n");
FileWriter writer = new FileWriter(file, true);
writer.write(java.time.LocalDate.now() + ", " +
java.time.LocalTime.now() + ": " + message + '\n');
writer.close();
} catch (IOException e) {
e.printStackTrace();
......
src/Dstore.java
+ 120
− 8
View file @ c7292a34
import java.io.*;
import java.net.*;
import java.nio.charset.StandardCharsets;
public class Dstore {
static File file = new File("output.txt");
public static void main (String[] args) {
// port to listen on
......@@ -15,22 +18,131 @@ public class Dstore {
String file_folder = args[3];
InetAddress ip;
createLogFile();
try {
ip = InetAddress.getLocalHost();
Socket socket = new Socket();
socket.connect(new InetSocketAddress(ip, cport), timeout);
socket.connect(new InetSocketAddress(ip, cport));
for(;;) {
PrintWriter out = new PrintWriter(socket.getOutputStream());
out.println("adding Dstore");
out.flush();
System.out.println("Dstore was added");
Thread.sleep(1000);
}
ServerSocket s = new ServerSocket(port);
Socket client = s.accept();
handleClient(client, socket, file_folder);
} catch (Exception e) {
System.out.println("error " + e);
}
}
private static void handleClient(Socket clientSocket, Socket controllerSocket, String file_folder) throws IOException, InterruptedException {
OutputStream outputStream = clientSocket.getOutputStream();
InputStream inputStream = clientSocket.getInputStream();
OutputStream cOutputStream = controllerSocket.getOutputStream();
InputStream cInputStream = controllerSocket.getInputStream();
BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));
String line;
while((line = reader.readLine()) != null) {
String[] tokens = line.split(" ");
String cmd = tokens[0];
if (cmd.equals("STORE")) {
// ACK the command from the client
outputStream.write("ACK\n".getBytes());
// read the filename, filesize and create the data bytes to be read
String filename = tokens[1];
int filesize = Integer.parseInt(tokens[2]);
// read data and store the file
byte[] data = new byte[filesize];
inputStream.readNBytes(data, 0, filesize);
FileOutputStream out = new FileOutputStream(file_folder + "/" + filename);
out.write(data);
out.flush();
out.close();
cOutputStream.write(("STORE_ACK " + filename).getBytes());
} else if (cmd.equals("LOAD_DATA")) {
String filename = tokens[1];
FileInputStream file = new FileInputStream(file_folder + "/" + filename);
int n;
while((n = file.read())!= -1) {
outputStream.write(n);
}
file.close();
} else if (cmd.equals("LIST")) {
} else if (cmd.equals("QUIT")) {
clientSocket.close();
break;
} else {
String msg = "unknown command " + line;
// outputStream.write(msg.getBytes());
log(msg);
}
}
if(!clientSocket.isClosed() || !controllerSocket.isClosed()) {
clientSocket.close();
controllerSocket.close();
}
}
private static void handleController(Socket controllerSocket) throws IOException {
OutputStream outputStream = controllerSocket.getOutputStream();
InputStream inputStream = controllerSocket.getInputStream();
BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));
String line;
while((line = reader.readLine()) != null) {
String[] tokens = line.split(" ");
String cmd = tokens[0];
if (cmd.equals("REMOVE")) {
} else if (cmd.equals("QUIT")) {
controllerSocket.close();
break;
} else {
String msg = "unknown command " + line;
// outputStream.write(msg.getBytes());
log(msg);
}
}
if(!controllerSocket.isClosed()) {
controllerSocket.close();
}
}
private static void createLogFile() {
try {
if(file.createNewFile()) {
System.out.println("File created");
} else {
System.out.println("File already exists");
}
} catch (IOException e) {
e.printStackTrace();
}
}
private static void log(String message) {
try {
FileWriter writer = new FileWriter(file, true);
writer.write(java.time.LocalDate.now() + ", " +
java.time.LocalTime.now() + ": " + message + '\n');
writer.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
\ No newline at end of file
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment