Add implementation from Assignment 1
This commit is contained in:
@@ -0,0 +1,62 @@
|
||||
package dslab;
|
||||
|
||||
import dslab.exception.MalformedInputException;
|
||||
|
||||
import java.util.Objects;
|
||||
|
||||
public class Email {
|
||||
private String username;
|
||||
private String domain;
|
||||
|
||||
public Email(String email) throws MalformedInputException {
|
||||
if (email.split("@").length != 2)
|
||||
throw new MalformedInputException("error email addresses must be of the form user@domain: " + email);
|
||||
if (email.split("@")[0].isBlank())
|
||||
throw new MalformedInputException("error email addresses must be of the form user@domain" + email);
|
||||
if (email.split("@")[1].isBlank())
|
||||
throw new MalformedInputException("error email addresses must be of the form user@domain" + email);
|
||||
|
||||
this.username = email.split("@")[0];
|
||||
this.domain = email.split("@")[1];
|
||||
}
|
||||
|
||||
public Email(String username, String domain) {
|
||||
this.username = username;
|
||||
this.domain = domain;
|
||||
}
|
||||
|
||||
public String getUsername() {
|
||||
return username;
|
||||
}
|
||||
|
||||
public void setUsername(String username) {
|
||||
this.username = username;
|
||||
}
|
||||
|
||||
public String getDomain() {
|
||||
return domain;
|
||||
}
|
||||
|
||||
public void setDomain(String domain) {
|
||||
this.domain = domain;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return username + '@' + domain;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object o) {
|
||||
if (this == o) return true;
|
||||
if (o == null || getClass() != o.getClass()) return false;
|
||||
Email email = (Email) o;
|
||||
return Objects.equals(getUsername(), email.getUsername()) &&
|
||||
Objects.equals(getDomain(), email.getDomain());
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
return Objects.hash(getUsername(), getDomain());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,93 @@
|
||||
package dslab;
|
||||
|
||||
import dslab.exception.MissingInputException;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
public class Message {
|
||||
private ArrayList<Email> to = new ArrayList<>();
|
||||
private Email from;
|
||||
private String subject = "";
|
||||
private String data = "";
|
||||
private Integer id;
|
||||
|
||||
public Message() {
|
||||
}
|
||||
|
||||
public Message(ArrayList<Email> to, Email from, String subject, String data) {
|
||||
this.to = to;
|
||||
this.from = from;
|
||||
this.subject = subject;
|
||||
this.data = data;
|
||||
}
|
||||
|
||||
public void allFieldsSet() throws MissingInputException {
|
||||
if (this.subject == null) this.subject = "";
|
||||
if (this.data == null) this.data = "";
|
||||
if (this.to.isEmpty()) throw new MissingInputException("error no receiver");
|
||||
if (this.from == null) throw new MissingInputException("error no sender");
|
||||
}
|
||||
|
||||
public void addTo(Email email) {
|
||||
to.add(email);
|
||||
}
|
||||
|
||||
public String printTo() {
|
||||
if (this.to.isEmpty())
|
||||
return null;
|
||||
return this.to.stream().map(Object::toString).collect(Collectors.joining(","));
|
||||
}
|
||||
|
||||
public ArrayList<Email> getTo() {
|
||||
return to;
|
||||
}
|
||||
|
||||
public void setTo(ArrayList<Email> to) {
|
||||
this.to = to;
|
||||
}
|
||||
|
||||
public Email getFrom() {
|
||||
return from;
|
||||
}
|
||||
|
||||
public void setFrom(Email from) {
|
||||
this.from = from;
|
||||
}
|
||||
|
||||
public String getSubject() {
|
||||
return subject;
|
||||
}
|
||||
|
||||
public void setSubject(String subject) {
|
||||
this.subject = subject;
|
||||
}
|
||||
|
||||
public String getData() {
|
||||
return data;
|
||||
}
|
||||
|
||||
public void setData(String data) {
|
||||
this.data = data;
|
||||
}
|
||||
|
||||
public Integer getId() {
|
||||
return id;
|
||||
}
|
||||
|
||||
public void setId(Integer id) {
|
||||
this.id = id;
|
||||
}
|
||||
|
||||
public String listMessage() {
|
||||
return getId() + " " + getFrom() + " " + getSubject();
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return "from " + getFrom().toString() + "\n" +
|
||||
"to " + printTo() + "\n" +
|
||||
"subject " + getSubject() + "\n" +
|
||||
"data " + getData() + "\n";
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
package dslab.exception;
|
||||
|
||||
public class MalformedInputException extends Exception {
|
||||
public MalformedInputException(String errorMessage) {
|
||||
super(errorMessage);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
package dslab.exception;
|
||||
|
||||
public class MessageNotFoundException extends Exception {
|
||||
public MessageNotFoundException(String errorMessage) {
|
||||
super(errorMessage);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
package dslab.exception;
|
||||
|
||||
public class MissingInputException extends Exception {
|
||||
public MissingInputException(String errorMessage) {
|
||||
super(errorMessage);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
package dslab.exception;
|
||||
|
||||
public class UnknownDomainException extends Exception {
|
||||
public UnknownDomainException(String errorMessage) {
|
||||
super(errorMessage);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
package dslab.exception;
|
||||
|
||||
public class UnknownRecipientException extends Exception {
|
||||
public UnknownRecipientException(String errorMessage) {
|
||||
super(errorMessage);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,194 @@
|
||||
package dslab.mailbox;
|
||||
|
||||
import dslab.Email;
|
||||
import dslab.Message;
|
||||
import dslab.exception.MessageNotFoundException;
|
||||
|
||||
import java.io.*;
|
||||
import java.net.Socket;
|
||||
import java.net.SocketException;
|
||||
import java.util.LinkedList;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.logging.Logger;
|
||||
|
||||
public class DMAPConnection implements Runnable {
|
||||
Logger logger = Logger.getLogger(DMAPConnection.class.getName());
|
||||
private final Socket socket;
|
||||
private PrintWriter out;
|
||||
private BufferedReader in;
|
||||
private Email currentUser = null;
|
||||
private final ConcurrentHashMap<Email, LinkedList<Message>> storage;
|
||||
private final ConcurrentHashMap<String, String> userStorage;
|
||||
|
||||
public DMAPConnection(Socket connection, ConcurrentHashMap<Email, LinkedList<Message>> storage, ConcurrentHashMap<String, String> userStorage) {
|
||||
this.socket = connection;
|
||||
this.storage = storage;
|
||||
this.userStorage = userStorage;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
logger.finer("Preparing for DMAP communication in " + this.toString());
|
||||
try {
|
||||
this.out = new PrintWriter(socket.getOutputStream(), true);
|
||||
this.in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
|
||||
out.println("ok DMAP");
|
||||
loginLoop();
|
||||
|
||||
String userInput;
|
||||
while (!Thread.currentThread().isInterrupted() && (userInput = in.readLine()) != null) {
|
||||
if ("quit".equals(userInput)) {
|
||||
out.println("ok bye");
|
||||
shutdown();
|
||||
} else if ("logout".equals(userInput)) {
|
||||
out.println("ok");
|
||||
currentUser = null;
|
||||
loginLoop();
|
||||
} else if ("list".equals(userInput)) {
|
||||
listMessages();
|
||||
} else if ("delete".equals(userInput.split("\\s+")[0])) {
|
||||
if (userInput.split("\\s+").length == 2) {
|
||||
try {
|
||||
deleteMessage(userInput.split("\\s+")[1]);
|
||||
} catch (MessageNotFoundException e) {
|
||||
out.println(e.getMessage());
|
||||
}
|
||||
} else {
|
||||
out.println("Please supply a message id to delete!");
|
||||
}
|
||||
} else if ("show".equals(userInput.split("\\s+")[0])) {
|
||||
if (userInput.split("\\s+").length == 2) {
|
||||
try {
|
||||
showMessage(userInput.split("\\s+")[1]);
|
||||
} catch (MessageNotFoundException e) {
|
||||
out.println(e.getMessage());
|
||||
}
|
||||
} else {
|
||||
out.println("Please supply a message id to show!");
|
||||
}
|
||||
} else {
|
||||
out.println("error protocol error");
|
||||
shutdown();
|
||||
}
|
||||
}
|
||||
} catch (InterruptedIOException ioe) {
|
||||
logger.info("Received interrupt from parent. Shutting down...");
|
||||
shutdown();
|
||||
} catch (IOException e) {
|
||||
logger.severe("Failed to get IO-Stream");
|
||||
e.printStackTrace();
|
||||
shutdown();
|
||||
}
|
||||
}
|
||||
|
||||
private void loginLoop() {
|
||||
String userInput;
|
||||
|
||||
try {
|
||||
while (!Thread.currentThread().isInterrupted()) {
|
||||
userInput = in.readLine();
|
||||
if ("quit".equals(userInput.split("\\s+")[0])) {
|
||||
out.println("ok bye");
|
||||
shutdown();
|
||||
} else if ("login".equals(userInput.split("\\s+")[0])) {
|
||||
String[] args = userInput.split("\\s+");
|
||||
if (args.length != 3)
|
||||
out.println("Please specify a username and password to login!");
|
||||
if (this.userStorage.containsKey(args[1])) {
|
||||
// Check if username exists
|
||||
if (args[2].equals(this.userStorage.get(args[1]))) {
|
||||
// Check if password matches
|
||||
for (Email email : this.storage.keySet()) {
|
||||
if (args[1].equals(email.getUsername())) {
|
||||
// Set current user if login successful
|
||||
currentUser = email;
|
||||
logger.info("User successfully logged in: " + currentUser.toString());
|
||||
out.println("ok");
|
||||
return;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
out.println("error wrong password");
|
||||
}
|
||||
} else {
|
||||
out.println("error unknown user");
|
||||
}
|
||||
} else {
|
||||
out.println("error not logged in");
|
||||
}
|
||||
}
|
||||
} catch (InterruptedIOException ioe) {
|
||||
logger.info("Received interrupt from parent. Shutting down...");
|
||||
shutdown();
|
||||
} catch (SocketException e) {
|
||||
logger.finer("Received interrupt. Exiting " + this.toString());
|
||||
shutdown();
|
||||
} catch (IOException e) {
|
||||
logger.severe("Failed to get IO-Stream");
|
||||
e.printStackTrace();
|
||||
shutdown();
|
||||
}
|
||||
}
|
||||
|
||||
private void showMessage(String id) throws MessageNotFoundException {
|
||||
int i;
|
||||
try {
|
||||
i = Integer.parseInt(id);
|
||||
} catch (NumberFormatException e) {
|
||||
throw new MessageNotFoundException("error unknown message id");
|
||||
}
|
||||
|
||||
for (Message m : storage.get(currentUser)) {
|
||||
if (m.getId() == i) {
|
||||
out.println(m);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
throw new MessageNotFoundException("error unknown message id");
|
||||
}
|
||||
|
||||
private void listMessages() {
|
||||
if (storage.get(currentUser).isEmpty()) {
|
||||
out.println("You do not have any messages at the moment!");
|
||||
return;
|
||||
}
|
||||
|
||||
for (Message m : storage.get(currentUser)) {
|
||||
out.println(m.listMessage());
|
||||
}
|
||||
}
|
||||
|
||||
public void deleteMessage (String id) throws MessageNotFoundException {
|
||||
int i;
|
||||
try {
|
||||
i = Integer.parseInt(id);
|
||||
} catch (NumberFormatException e) {
|
||||
throw new MessageNotFoundException("error unknown message id");
|
||||
}
|
||||
|
||||
for (Message m : storage.get(currentUser)) {
|
||||
if (m.getId() == i) {
|
||||
storage.get(currentUser).remove(m);
|
||||
out.println("ok");
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
throw new MessageNotFoundException("error unknown message id");
|
||||
}
|
||||
|
||||
public void shutdown() {
|
||||
logger.info("Shutting down client connection " + this.toString());
|
||||
try {
|
||||
if (socket != null)
|
||||
socket.close();
|
||||
in.close();
|
||||
out.close();
|
||||
} catch (IOException e) {
|
||||
logger.severe("Error closing socket and/or IO-streams");
|
||||
e.printStackTrace();
|
||||
}
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,73 @@
|
||||
package dslab.mailbox;
|
||||
|
||||
import dslab.Email;
|
||||
import dslab.Message;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.io.InterruptedIOException;
|
||||
import java.net.ServerSocket;
|
||||
import java.net.Socket;
|
||||
import java.net.SocketException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.LinkedList;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.logging.Logger;
|
||||
|
||||
public class DMAPListener extends Thread {
|
||||
private final ServerSocket serverSocket;
|
||||
private final Logger logger = Logger.getLogger(DMAPListener.class.getName());
|
||||
private final ConcurrentHashMap<Email, LinkedList<Message>> storage;
|
||||
private final ConcurrentHashMap<String, String> userStorage;
|
||||
private final ArrayList<DMAPConnection> clients = new ArrayList<>();
|
||||
private final ExecutorService executorService = Executors.newCachedThreadPool();
|
||||
|
||||
public DMAPListener(ServerSocket serverSocket, ConcurrentHashMap<Email, LinkedList<Message>> storage, ConcurrentHashMap<String, String> userStorage) {
|
||||
this.serverSocket = serverSocket;
|
||||
this.storage = storage;
|
||||
this.userStorage = userStorage;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
while (!Thread.currentThread().isInterrupted()) {
|
||||
logger.finer("Waiting for request on serverSocket " + serverSocket.toString());
|
||||
try {
|
||||
Socket s = serverSocket.accept();
|
||||
logger.fine("Processing incoming socket " + s.toString());
|
||||
DMAPConnection dmapConnection = new DMAPConnection(s, storage, userStorage);
|
||||
clients.add(dmapConnection);
|
||||
executorService.submit(dmapConnection);
|
||||
} catch (InterruptedIOException | SocketException e) {
|
||||
logger.finer("Received interrupt. Exiting " + this.toString());
|
||||
this.shutdown();
|
||||
} catch (IOException e) {
|
||||
logger.severe("Error starting serverSocket " + serverSocket.toString());
|
||||
e.printStackTrace();
|
||||
this.shutdown();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public void shutdown() {
|
||||
logger.finer("Shutting down DMAPListener " + this.toString());
|
||||
for (DMAPConnection client : clients) {
|
||||
if (client != null)
|
||||
client.shutdown();
|
||||
}
|
||||
executorService.shutdown();
|
||||
try {
|
||||
if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
|
||||
executorService.shutdownNow();
|
||||
if (!executorService.awaitTermination(60, TimeUnit.SECONDS))
|
||||
System.err.println("Pool did not terminate");
|
||||
}
|
||||
} catch (InterruptedException ie) {
|
||||
executorService.shutdownNow();
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
this.interrupt();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,166 @@
|
||||
package dslab.mailbox;
|
||||
|
||||
import dslab.Email;
|
||||
import dslab.Message;
|
||||
import dslab.exception.MalformedInputException;
|
||||
|
||||
import java.io.*;
|
||||
import java.net.Socket;
|
||||
import java.net.SocketException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.LinkedList;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.logging.Logger;
|
||||
|
||||
public class DMTPConnection implements Runnable {
|
||||
Logger logger = Logger.getLogger(DMTPConnection.class.getName());
|
||||
private final Socket socket;
|
||||
private PrintWriter out;
|
||||
private BufferedReader in;
|
||||
|
||||
private Message msg = new Message();
|
||||
|
||||
private final String domain;
|
||||
|
||||
private final ConcurrentHashMap<Email, LinkedList<Message>> messageStorage;
|
||||
private final ConcurrentHashMap<String, String> userStorage;
|
||||
|
||||
public DMTPConnection(Socket connection, ConcurrentHashMap<Email, LinkedList<Message>> messageStorage, ConcurrentHashMap<String, String> userStorage, String domain) {
|
||||
this.socket = connection;
|
||||
this.messageStorage = messageStorage;
|
||||
this.userStorage = userStorage;
|
||||
this.domain = domain;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
logger.finer("Preparing for DMTP communication in " + this.toString());
|
||||
try {
|
||||
this.out = new PrintWriter(socket.getOutputStream(), true);
|
||||
this.in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
|
||||
out.println("ok DMTP");
|
||||
|
||||
String userInput;
|
||||
|
||||
try {
|
||||
if (!("begin".equals(in.readLine()))) {
|
||||
out.println("error protocol error");
|
||||
shutdown();
|
||||
}
|
||||
} catch (SocketException e) {
|
||||
// Thrown if socket has already been closed by shutdown() method
|
||||
logger.finer("Received interrupt. Exiting " + this.toString());
|
||||
shutdown();
|
||||
}
|
||||
out.println("ok");
|
||||
|
||||
while (!Thread.currentThread().isInterrupted() && (userInput = in.readLine()) != null) {
|
||||
if ("quit".equals(userInput)) {
|
||||
out.println("ok bye");
|
||||
shutdown();
|
||||
} else if ("send".equals(userInput)) {
|
||||
storeMessage();
|
||||
out.println("ok");
|
||||
} else if ("to".equals(userInput.split("\\s+")[0])) {
|
||||
msg.setTo(new ArrayList<>());
|
||||
String[] emailAddresses = userInput.split("\\s+")[1].split(",");
|
||||
logger.info("Split EMail addresses: " + Arrays.toString(emailAddresses));
|
||||
int count = 0;
|
||||
try {
|
||||
for (String emailAddress : emailAddresses) {
|
||||
logger.info("Current email address in msg.getTo(): " + emailAddress);
|
||||
Email add = new Email(emailAddress);
|
||||
if (this.domain.equals(add.getDomain())) {
|
||||
logger.info("Address " + emailAddress + " belongs to this domain " + this.domain);
|
||||
if (!this.userStorage.containsKey(add.getUsername())) {
|
||||
logger.info("Our userStorage in domain " + this.domain + " does not contain user " + add.getUsername());
|
||||
out.println("error unknown recipient " + add.toString());
|
||||
} else {
|
||||
logger.info("Address " + add.toString() + " belongs to this domain and user exists. Adding address to msg.To() field");
|
||||
msg.addTo(add);
|
||||
count++;
|
||||
}
|
||||
} else {
|
||||
logger.info("Address " + emailAddress + " does not belong to this domain. Adding anyway...");
|
||||
msg.addTo(add);
|
||||
count++;
|
||||
}
|
||||
}
|
||||
if (count != 0)
|
||||
out.println("ok " + count);
|
||||
} catch (MalformedInputException e) {
|
||||
out.println(e.getMessage());
|
||||
}
|
||||
} else if ("from".equals(userInput.split("\\s+")[0])) {
|
||||
try {
|
||||
Email from = new Email(userInput.split("\\s+")[1]);
|
||||
this.msg.setFrom(from);
|
||||
out.println("ok");
|
||||
} catch (MalformedInputException mie) {
|
||||
out.println(mie.getMessage());
|
||||
}
|
||||
} else if ("subject".equals(userInput.split("\\s+")[0])) {
|
||||
String subject = "";
|
||||
if (userInput.split("\\s+").length > 1)
|
||||
subject = userInput.split("\\s+", 2)[1];
|
||||
msg.setSubject(subject);
|
||||
out.println("ok");
|
||||
} else if ("data".equals(userInput.split("\\s+")[0])) {
|
||||
String data = userInput.split("\\s+", 2)[1];
|
||||
msg.setData(data);
|
||||
out.println("ok");
|
||||
} else {
|
||||
out.println("error protocol error");
|
||||
shutdown();
|
||||
}
|
||||
}
|
||||
} catch (InterruptedIOException ioe) {
|
||||
logger.info("Received interrupt from parent. Shutting down...");
|
||||
shutdown();
|
||||
} catch (IOException e) {
|
||||
logger.severe("Failed to get IO-Stream");
|
||||
e.printStackTrace();
|
||||
shutdown();
|
||||
}
|
||||
}
|
||||
|
||||
private synchronized void storeMessage() {
|
||||
logger.info("Storing message " + msg.toString());
|
||||
this.msg.setId(MailboxServer.id++);
|
||||
boolean alreadyPresent;
|
||||
for (Email recipient : this.msg.getTo()) {
|
||||
alreadyPresent = false;
|
||||
logger.info("storeMessage(): checking if msg " + msg.listMessage() + " already exists for recipient " + recipient.toString());
|
||||
if (this.messageStorage.containsKey(recipient)) {
|
||||
// Check if message already exists for user
|
||||
for (Message m : this.messageStorage.get(recipient)) {
|
||||
if (this.msg.getId().equals(m.getId())) {
|
||||
logger.info("storeMessage(): msg " + msg.listMessage() + " already exists for recipient " + recipient.toString());
|
||||
alreadyPresent = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
// Save only if message doesn't already exist
|
||||
if (!alreadyPresent) {
|
||||
this.messageStorage.get(recipient).add(this.msg);
|
||||
}
|
||||
}
|
||||
}
|
||||
this.msg = new Message();
|
||||
}
|
||||
|
||||
public void shutdown() {
|
||||
logger.info("Shutting down client connection " + this.toString());
|
||||
try {
|
||||
if (socket != null)
|
||||
socket.close();
|
||||
in.close();
|
||||
out.close();
|
||||
} catch (IOException e) {
|
||||
logger.severe("Error closing socket and/or IO-streams");
|
||||
e.printStackTrace();
|
||||
}
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,75 @@
|
||||
package dslab.mailbox;
|
||||
|
||||
import dslab.Email;
|
||||
import dslab.Message;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.io.InterruptedIOException;
|
||||
import java.net.ServerSocket;
|
||||
import java.net.Socket;
|
||||
import java.net.SocketException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.LinkedList;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.logging.Logger;
|
||||
|
||||
public class DMTPListener extends Thread {
|
||||
private final ServerSocket serverSocket;
|
||||
private final Logger logger = Logger.getLogger(DMTPListener.class.getName());
|
||||
private final ArrayList<DMTPConnection> clients = new ArrayList<>();
|
||||
private final ExecutorService executorService = Executors.newCachedThreadPool();
|
||||
private final ConcurrentHashMap<Email, LinkedList<Message>> messageStorage;
|
||||
private final ConcurrentHashMap<String, String> userStorage;
|
||||
private final String domain;
|
||||
|
||||
public DMTPListener(ServerSocket serverSocket, ConcurrentHashMap<Email, LinkedList<Message>> storage, ConcurrentHashMap<String, String> userStorage, String domain) {
|
||||
this.serverSocket = serverSocket;
|
||||
this.messageStorage = storage;
|
||||
this.userStorage = userStorage;
|
||||
this.domain = domain;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
while (!Thread.currentThread().isInterrupted()) {
|
||||
logger.finer("Waiting for request on serverSocket " + serverSocket.toString());
|
||||
try {
|
||||
Socket s = serverSocket.accept();
|
||||
logger.fine("Processing incoming socket " + s.toString());
|
||||
DMTPConnection dmtpConnection = new DMTPConnection(s, messageStorage, userStorage, domain);
|
||||
clients.add(dmtpConnection);
|
||||
executorService.submit(dmtpConnection);
|
||||
} catch (InterruptedIOException | SocketException e) {
|
||||
logger.finer("Received interrupt. Exiting " + this.toString());
|
||||
this.shutdown();
|
||||
} catch (IOException e) {
|
||||
logger.severe("Error starting serverSocket " + serverSocket.toString());
|
||||
e.printStackTrace();
|
||||
this.shutdown();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public void shutdown() {
|
||||
logger.finer("Shutting down DMTPListener " + this.toString());
|
||||
for (DMTPConnection client : clients) {
|
||||
if (client != null)
|
||||
client.shutdown();
|
||||
}
|
||||
executorService.shutdown();
|
||||
try {
|
||||
if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
|
||||
executorService.shutdownNow();
|
||||
if (!executorService.awaitTermination(60, TimeUnit.SECONDS))
|
||||
System.err.println("Pool did not terminate");
|
||||
}
|
||||
} catch (InterruptedException ie) {
|
||||
executorService.shutdownNow();
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
this.interrupt();
|
||||
}
|
||||
}
|
||||
@@ -1,12 +1,36 @@
|
||||
package dslab.mailbox;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.io.PrintStream;
|
||||
import java.net.ServerSocket;
|
||||
import java.util.LinkedList;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.logging.Logger;
|
||||
|
||||
import at.ac.tuwien.dsg.orvell.Shell;
|
||||
import at.ac.tuwien.dsg.orvell.StopShellException;
|
||||
import at.ac.tuwien.dsg.orvell.annotation.Command;
|
||||
import dslab.ComponentFactory;
|
||||
import dslab.Email;
|
||||
import dslab.Message;
|
||||
import dslab.util.Config;
|
||||
|
||||
public class MailboxServer implements IMailboxServer, Runnable {
|
||||
private static final Logger logger = Logger.getLogger(MailboxServer.class.getName());
|
||||
private final String domain;
|
||||
private ServerSocket dmtpServerSocket;
|
||||
private ServerSocket dmapServerSocket;
|
||||
private final Shell shell;
|
||||
private final Integer dmtpServerPort;
|
||||
private final Integer dmapServerPort;
|
||||
private DMTPListener dmtpListener;
|
||||
private DMAPListener dmapListener;
|
||||
private final ConcurrentHashMap<Email, LinkedList<Message>> messageStorage = new ConcurrentHashMap<>();
|
||||
private final ConcurrentHashMap<String, String> userStorage = new ConcurrentHashMap<>();
|
||||
|
||||
public static volatile Integer id;
|
||||
|
||||
|
||||
/**
|
||||
* Creates a new server instance.
|
||||
@@ -17,17 +41,67 @@ public class MailboxServer implements IMailboxServer, Runnable {
|
||||
* @param out the output stream to write console output to
|
||||
*/
|
||||
public MailboxServer(String componentId, Config config, InputStream in, PrintStream out) {
|
||||
// TODO
|
||||
this.domain = config.getString("domain");
|
||||
Config userConfig = new Config(config.getString("users.config"));
|
||||
|
||||
// Load users from config into userStorage for authentication
|
||||
for (String key : userConfig.listKeys()) {
|
||||
userStorage.put(key, userConfig.getString(key));
|
||||
}
|
||||
// Load Email Addresses into messageStorage
|
||||
for (String key : userStorage.keySet()) {
|
||||
Email current = new Email(key, domain);
|
||||
messageStorage.put(current, new LinkedList<>());
|
||||
}
|
||||
|
||||
id = 0;
|
||||
|
||||
this.shell = new Shell(in, out);
|
||||
this.shell.register(this);
|
||||
this.shell.setPrompt("Mailboxserver> ");
|
||||
this.dmtpServerPort = config.getInt("dmtp.tcp.port");
|
||||
this.dmapServerPort = config.getInt("dmap.tcp.port");
|
||||
}
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
// TODO
|
||||
logger.info("Creating DMTP and DMAP serverSockets for MailboxServer + " + this.toString());
|
||||
try {
|
||||
this.dmtpServerSocket = new ServerSocket(dmtpServerPort);
|
||||
this.dmapServerSocket = new ServerSocket(dmapServerPort);
|
||||
} catch (IOException e) {
|
||||
logger.severe("Error creating one of the two server sockets");
|
||||
e.printStackTrace();
|
||||
shutdown();
|
||||
}
|
||||
this.dmtpListener = new DMTPListener(this.dmtpServerSocket, this.messageStorage, this.userStorage, this.domain);
|
||||
this.dmtpListener.start();
|
||||
this.dmapListener = new DMAPListener(this.dmapServerSocket, this.messageStorage, this.userStorage);
|
||||
this.dmapListener.start();
|
||||
this.shell.run();
|
||||
}
|
||||
|
||||
@Command
|
||||
@Override
|
||||
public void shutdown() {
|
||||
// TODO
|
||||
try {
|
||||
if (dmtpServerSocket != null)
|
||||
dmtpServerSocket.close();
|
||||
this.dmtpListener.interrupt();
|
||||
} catch (IOException e) {
|
||||
logger.severe("Error closing DMTP serverSocket " + dmtpServerSocket.toString());
|
||||
e.printStackTrace();
|
||||
}
|
||||
|
||||
try {
|
||||
if (dmapServerSocket != null)
|
||||
dmapServerSocket.close();
|
||||
this.dmapListener.interrupt();
|
||||
} catch (IOException e) {
|
||||
logger.severe("Error closing DMTP serverSocket " + dmapServerSocket.toString());
|
||||
e.printStackTrace();
|
||||
}
|
||||
throw new StopShellException();
|
||||
}
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
|
||||
@@ -0,0 +1,64 @@
|
||||
package dslab.monitoring;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.io.InterruptedIOException;
|
||||
import java.net.DatagramPacket;
|
||||
import java.net.DatagramSocket;
|
||||
import java.net.SocketException;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.logging.Logger;
|
||||
|
||||
public class MonitoringListener extends Thread {
|
||||
private static final Logger logger = Logger.getLogger(MonitoringListener.class.getName());
|
||||
private final DatagramSocket serverSocket;
|
||||
private final ConcurrentHashMap<String, Integer> addresses;
|
||||
private final ConcurrentHashMap<String, Integer> servers;
|
||||
|
||||
public MonitoringListener(DatagramSocket serverSocket, ConcurrentHashMap<String, Integer> addresses, ConcurrentHashMap<String, Integer> servers) {
|
||||
this.serverSocket = serverSocket;
|
||||
this.addresses = addresses;
|
||||
this.servers = servers;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
byte[] recvBuffer = new byte[1024];
|
||||
while (!Thread.currentThread().isInterrupted()) {
|
||||
logger.finer("Waiting for request on UDP serverSocket " + serverSocket.toString());
|
||||
try {
|
||||
DatagramPacket packet = new DatagramPacket(recvBuffer, recvBuffer.length);
|
||||
serverSocket.receive(packet);
|
||||
logger.fine("Processing incoming socket " + serverSocket.toString());
|
||||
String receivedData = new String(packet.getData(), StandardCharsets.UTF_8).trim();
|
||||
String server = receivedData.split("\\s+")[0];
|
||||
String address = receivedData.split("\\s+")[1];
|
||||
logger.info("Incoming packet contains server: " + server);
|
||||
logger.info("Incoming packet contains address: " + address);
|
||||
// Add/Update num of packets sent by server
|
||||
if (!this.servers.containsKey(server))
|
||||
this.servers.put(server, 1);
|
||||
else
|
||||
this.servers.put(server, this.servers.get(server) + 1);
|
||||
// Add/Update num of packets sent for address
|
||||
if (!this.addresses.containsKey(address))
|
||||
this.addresses.put(address, 1);
|
||||
else
|
||||
this.addresses.put(address, this.addresses.get(address) + 1);
|
||||
} catch (InterruptedIOException | SocketException e) {
|
||||
logger.finer("Received interrupt. Exiting " + this.toString());
|
||||
this.shutdown();
|
||||
} catch (IOException e) {
|
||||
logger.severe("Error starting serverSocket " + serverSocket.toString());
|
||||
e.printStackTrace();
|
||||
this.shutdown();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public void shutdown() {
|
||||
logger.finer("Shutting down MonitoringListener " + this.toString());
|
||||
this.serverSocket.close();
|
||||
this.interrupt();
|
||||
}
|
||||
}
|
||||
@@ -1,12 +1,27 @@
|
||||
package dslab.monitoring;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.io.PrintStream;
|
||||
import java.net.DatagramSocket;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.logging.Logger;
|
||||
|
||||
import at.ac.tuwien.dsg.orvell.Shell;
|
||||
import at.ac.tuwien.dsg.orvell.StopShellException;
|
||||
import at.ac.tuwien.dsg.orvell.annotation.Command;
|
||||
import dslab.ComponentFactory;
|
||||
import dslab.util.Config;
|
||||
|
||||
public class MonitoringServer implements IMonitoringServer {
|
||||
private static final Logger logger = Logger.getLogger(MonitoringServer.class.getName());
|
||||
private DatagramSocket serverSocket;
|
||||
private final Shell shell;
|
||||
private final Integer serverPort;
|
||||
private MonitoringListener monitoringListener;
|
||||
private final ConcurrentHashMap<String, Integer> addresses = new ConcurrentHashMap<>();
|
||||
private final ConcurrentHashMap<String, Integer> servers = new ConcurrentHashMap<>();
|
||||
|
||||
|
||||
/**
|
||||
* Creates a new server instance.
|
||||
@@ -17,27 +32,50 @@ public class MonitoringServer implements IMonitoringServer {
|
||||
* @param out the output stream to write console output to
|
||||
*/
|
||||
public MonitoringServer(String componentId, Config config, InputStream in, PrintStream out) {
|
||||
// TODO
|
||||
this.shell = new Shell(in, out);
|
||||
this.shell.register(this);
|
||||
this.shell.setPrompt("MonitoringServer> ");
|
||||
this.serverPort = config.getInt("udp.port");
|
||||
}
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
// TODO
|
||||
logger.info("Creating UDP serverSocket for " + this.toString());
|
||||
try {
|
||||
this.serverSocket = new DatagramSocket(serverPort);
|
||||
} catch (IOException e) {
|
||||
logger.severe("Error creating serverSocket " + serverSocket.toString());
|
||||
e.printStackTrace();
|
||||
shutdown();
|
||||
}
|
||||
this.monitoringListener = new MonitoringListener(serverSocket, addresses, servers);
|
||||
this.monitoringListener.start();
|
||||
this.shell.run();
|
||||
}
|
||||
|
||||
@Command
|
||||
@Override
|
||||
public void addresses() {
|
||||
// TODO
|
||||
for (String address : addresses.keySet()) {
|
||||
this.shell.out().println(address + " " + addresses.get(address));
|
||||
}
|
||||
}
|
||||
|
||||
@Command
|
||||
@Override
|
||||
public void servers() {
|
||||
// TODO
|
||||
for (String server : servers.keySet()) {
|
||||
this.shell.out().println(server + " " + servers.get(server));
|
||||
}
|
||||
}
|
||||
|
||||
@Command
|
||||
@Override
|
||||
public void shutdown() {
|
||||
// TODO
|
||||
if (serverSocket != null)
|
||||
serverSocket.close();
|
||||
this.monitoringListener.interrupt();
|
||||
throw new StopShellException();
|
||||
}
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
|
||||
@@ -0,0 +1,129 @@
|
||||
package dslab.transfer;
|
||||
|
||||
import dslab.Email;
|
||||
import dslab.Message;
|
||||
import dslab.exception.MalformedInputException;
|
||||
import dslab.exception.MissingInputException;
|
||||
|
||||
import java.io.*;
|
||||
import java.net.Socket;
|
||||
import java.net.SocketException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.logging.Logger;
|
||||
|
||||
public class ClientConnection implements Runnable {
|
||||
Logger logger = Logger.getLogger(ClientConnection.class.getName());
|
||||
private final Socket socket;
|
||||
private PrintWriter out;
|
||||
private BufferedReader in;
|
||||
|
||||
private Message msg = new Message();
|
||||
|
||||
public ClientConnection(Socket connection) {
|
||||
this.socket = connection;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
logger.finer("Preparing for DMTP communication in " + this.toString());
|
||||
try {
|
||||
this.out = new PrintWriter(socket.getOutputStream(), true);
|
||||
this.in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
|
||||
out.println("ok DMTP");
|
||||
|
||||
String userInput;
|
||||
|
||||
try {
|
||||
if (!("begin".equals(in.readLine()))) {
|
||||
out.println("error protocol error");
|
||||
shutdown();
|
||||
}
|
||||
} catch (SocketException e) {
|
||||
// Thrown if socket has already been closed by shutdown() method
|
||||
// TODO maybe send message that connection has been closed by server?
|
||||
logger.finer("Received interrupt. Exiting " + this.toString());
|
||||
shutdown();
|
||||
}
|
||||
out.println("ok");
|
||||
|
||||
while (!Thread.currentThread().isInterrupted() && (userInput = in.readLine()) != null) {
|
||||
if ("quit".equals(userInput)) {
|
||||
out.println("ok bye");
|
||||
shutdown();
|
||||
} else if ("send".equals(userInput)) {
|
||||
try {
|
||||
sendMessage();
|
||||
out.println("ok");
|
||||
} catch (MissingInputException e) {
|
||||
out.println(e.getMessage());
|
||||
}
|
||||
} else if ("to".equals(userInput.split("\\s+")[0])) {
|
||||
msg.setTo(new ArrayList<>());
|
||||
String[] emailAddresses = userInput.split("\\s+")[1].split(",");
|
||||
int count = 0;
|
||||
try {
|
||||
for (String emailAddress : emailAddresses) {
|
||||
msg.addTo(new Email(emailAddress));
|
||||
count++;
|
||||
}
|
||||
out.println("ok " + count);
|
||||
} catch (MalformedInputException mie) {
|
||||
out.println(mie.getMessage());
|
||||
}
|
||||
} else if ("from".equals(userInput.split("\\s+")[0])) {
|
||||
try {
|
||||
Email from = new Email(userInput.split("\\s+")[1]);
|
||||
this.msg.setFrom(from);
|
||||
out.println("ok");
|
||||
} catch (MalformedInputException mie) {
|
||||
out.println(mie.getMessage());
|
||||
}
|
||||
} else if ("subject".equals(userInput.split("\\s+")[0])) {
|
||||
String subject = "";
|
||||
if (userInput.split("\\s+").length > 1)
|
||||
subject = userInput.split("\\s+", 2)[1];
|
||||
logger.info("Setting subject to: " + subject);
|
||||
msg.setSubject(subject);
|
||||
out.println("ok");
|
||||
} else if ("data".equals(userInput.split("\\s+")[0])) {
|
||||
String data = "";
|
||||
if (userInput.split("\\s+").length > 1)
|
||||
data = userInput.split("\\s+", 2)[1];
|
||||
logger.info("Setting data to: " + data);
|
||||
msg.setData(data);
|
||||
out.println("ok");
|
||||
} else {
|
||||
out.println("error protocol error");
|
||||
shutdown();
|
||||
}
|
||||
}
|
||||
} catch (InterruptedIOException ioe) {
|
||||
logger.info("Received interrupt from parent. Shutting down...");
|
||||
shutdown();
|
||||
} catch (IOException e) {
|
||||
logger.severe("Failed to get IO-Stream");
|
||||
shutdown();
|
||||
}
|
||||
}
|
||||
|
||||
public void shutdown() {
|
||||
logger.info("Shutting down client connection " + this.toString());
|
||||
try {
|
||||
if (socket != null)
|
||||
socket.close();
|
||||
in.close();
|
||||
out.close();
|
||||
} catch (IOException e) {
|
||||
logger.severe("Error closing socket and/or IO-streams");
|
||||
e.printStackTrace();
|
||||
}
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
|
||||
public void sendMessage() throws MissingInputException {
|
||||
this.msg.allFieldsSet();
|
||||
TransferServer.Producer producer = new TransferServer.Producer(this.msg);
|
||||
new Thread(producer).start();
|
||||
this.msg = new Message();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,64 @@
|
||||
package dslab.transfer;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.io.InterruptedIOException;
|
||||
import java.net.ServerSocket;
|
||||
import java.net.Socket;
|
||||
import java.net.SocketException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.logging.Logger;
|
||||
|
||||
public class ClientListener extends Thread {
|
||||
private final ServerSocket serverSocket;
|
||||
private final Logger logger = Logger.getLogger(ClientListener.class.getName());
|
||||
private final ArrayList<ClientConnection> clients = new ArrayList<>();
|
||||
private final ExecutorService executorService = Executors.newCachedThreadPool();
|
||||
|
||||
public ClientListener(ServerSocket serverSocket) {
|
||||
this.serverSocket = serverSocket;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
while (!Thread.currentThread().isInterrupted()) {
|
||||
logger.finer("Waiting for request on serverSocket " + serverSocket.toString());
|
||||
try {
|
||||
Socket s = serverSocket.accept();
|
||||
logger.fine("Processing incoming socket " + s.toString());
|
||||
ClientConnection clientConnection = new ClientConnection(s);
|
||||
clients.add(clientConnection);
|
||||
executorService.submit(clientConnection);
|
||||
} catch (InterruptedIOException | SocketException e) {
|
||||
logger.finer("Received interrupt. Exiting " + this.toString());
|
||||
this.shutdown();
|
||||
} catch (IOException e) {
|
||||
logger.severe("Error starting serverSocket " + serverSocket.toString());
|
||||
e.printStackTrace();
|
||||
this.shutdown();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public void shutdown() {
|
||||
logger.finer("Shutting down ClientListener " + this.toString());
|
||||
for (ClientConnection client : clients) {
|
||||
if (client != null)
|
||||
client.shutdown();
|
||||
}
|
||||
executorService.shutdown();
|
||||
try {
|
||||
if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
|
||||
executorService.shutdownNow();
|
||||
if (!executorService.awaitTermination(60, TimeUnit.SECONDS))
|
||||
System.err.println("Pool did not terminate");
|
||||
}
|
||||
} catch (InterruptedException ie) {
|
||||
executorService.shutdownNow();
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
this.interrupt();
|
||||
}
|
||||
}
|
||||
@@ -1,12 +1,32 @@
|
||||
package dslab.transfer;
|
||||
|
||||
import java.io.InputStream;
|
||||
import java.io.PrintStream;
|
||||
import java.io.*;
|
||||
import java.net.*;
|
||||
import java.util.ArrayList;
|
||||
import java.util.HashMap;
|
||||
import java.util.concurrent.BlockingQueue;
|
||||
import java.util.concurrent.LinkedBlockingQueue;
|
||||
import java.util.logging.Logger;
|
||||
|
||||
import at.ac.tuwien.dsg.orvell.Shell;
|
||||
import at.ac.tuwien.dsg.orvell.StopShellException;
|
||||
import at.ac.tuwien.dsg.orvell.annotation.Command;
|
||||
import dslab.ComponentFactory;
|
||||
import dslab.Email;
|
||||
import dslab.Message;
|
||||
import dslab.exception.MalformedInputException;
|
||||
import dslab.exception.UnknownDomainException;
|
||||
import dslab.util.Config;
|
||||
|
||||
public class TransferServer implements ITransferServer, Runnable {
|
||||
private static final Logger logger = Logger.getLogger(TransferServer.class.getName());
|
||||
private ServerSocket serverSocket;
|
||||
private final Shell shell;
|
||||
private final Integer serverPort;
|
||||
public static final HashMap<String, Integer> mailboxServers = new HashMap<>();
|
||||
public static volatile BlockingQueue<Message> blockingQueue = new LinkedBlockingQueue<>(10);
|
||||
private final Consumer consumer;
|
||||
public static final Object lock = new Object();
|
||||
|
||||
/**
|
||||
* Creates a new server instance.
|
||||
@@ -17,17 +37,46 @@ public class TransferServer implements ITransferServer, Runnable {
|
||||
* @param out the output stream to write console output to
|
||||
*/
|
||||
public TransferServer(String componentId, Config config, InputStream in, PrintStream out) {
|
||||
// TODO
|
||||
Config univerze = new Config("mailbox-univer-ze.properties");
|
||||
Config earthplanet = new Config("mailbox-earth-planet.properties");
|
||||
mailboxServers.put(univerze.getString("domain"), univerze.getInt("dmtp.tcp.port"));
|
||||
mailboxServers.put(earthplanet.getString("domain"), earthplanet.getInt("dmtp.tcp.port"));
|
||||
String monitoringHost = config.getString("monitoring.host");
|
||||
Integer monitoringPort = config.getInt("monitoring.port");
|
||||
this.serverPort = config.getInt("tcp.port");
|
||||
this.consumer = new Consumer(mailboxServers, monitoringHost, monitoringPort, "127.0.0.1", this.serverPort);
|
||||
this.shell = new Shell(in, out);
|
||||
this.shell.register(this);
|
||||
this.shell.setPrompt("Transferserver> ");
|
||||
}
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
// TODO
|
||||
logger.info("Creating serverSocket for " + this.toString());
|
||||
try {
|
||||
this.serverSocket = new ServerSocket(serverPort);
|
||||
} catch (IOException e) {
|
||||
logger.severe("Error creating serverSocket " + serverSocket.toString());
|
||||
e.printStackTrace();
|
||||
shutdown();
|
||||
}
|
||||
new ClientListener(serverSocket).start();
|
||||
this.consumer.start();
|
||||
this.shell.run();
|
||||
}
|
||||
|
||||
@Command
|
||||
@Override
|
||||
public void shutdown() {
|
||||
// TODO
|
||||
try {
|
||||
if (serverSocket != null)
|
||||
serverSocket.close();
|
||||
} catch (IOException e) {
|
||||
logger.severe("Error closing serverSocket " + serverSocket.toString());
|
||||
e.printStackTrace();
|
||||
}
|
||||
this.consumer.interrupt();
|
||||
throw new StopShellException();
|
||||
}
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
@@ -35,4 +84,206 @@ public class TransferServer implements ITransferServer, Runnable {
|
||||
server.run();
|
||||
}
|
||||
|
||||
static class Producer extends Thread {
|
||||
private final Message msg;
|
||||
|
||||
Producer(Message msg) {
|
||||
this.msg = msg;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
try {
|
||||
produce();
|
||||
} catch (InterruptedException e) {
|
||||
logger.info("Producer thread has been interrupted. Exiting..." + this.toString());
|
||||
shutdown();
|
||||
}
|
||||
}
|
||||
|
||||
private void produce() throws InterruptedException {
|
||||
synchronized (lock) {
|
||||
while (blockingQueue.size() == 10) {
|
||||
lock.wait();
|
||||
}
|
||||
blockingQueue.put(msg);
|
||||
logger.info("Added message " + msg.toString() + " to queue");
|
||||
lock.notify();
|
||||
}
|
||||
}
|
||||
|
||||
private void shutdown() {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
|
||||
static class Consumer extends Thread {
|
||||
private final HashMap<String, Integer> mailboxServers;
|
||||
private final String monitoringHost;
|
||||
private final Integer monitoringPort;
|
||||
private final String transferHost;
|
||||
private final Integer transferPort;
|
||||
|
||||
Consumer(HashMap<String, Integer> mailboxServers,
|
||||
String monitoringHost,
|
||||
Integer monitoringPort,
|
||||
String transferHost,
|
||||
Integer transferPort) {
|
||||
this.mailboxServers = mailboxServers;
|
||||
this.monitoringHost = monitoringHost;
|
||||
this.monitoringPort = monitoringPort;
|
||||
this.transferHost = transferHost;
|
||||
this.transferPort = transferPort;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
try {
|
||||
consume();
|
||||
} catch (InterruptedException e) {
|
||||
logger.info("Consumer thread has been interrupted. Exiting..." + this.toString());
|
||||
shutdown();
|
||||
}
|
||||
}
|
||||
|
||||
private void consume() throws InterruptedException {
|
||||
while (!Thread.currentThread().isInterrupted()) {
|
||||
synchronized (lock) {
|
||||
while (blockingQueue.isEmpty()) {
|
||||
logger.info("Queue currently empty. Waiting for messages to appear...");
|
||||
lock.wait();
|
||||
}
|
||||
logger.info("Queue not empty. Processing message...");
|
||||
Message msg = blockingQueue.take();
|
||||
logger.info("Took message " + msg.toString() + " from queue");
|
||||
HashMap<String, Boolean> sent = new HashMap<>();
|
||||
sent.put("earth.planet", false);
|
||||
sent.put("univer.ze", false);
|
||||
for (Email recipient : msg.getTo()) {
|
||||
logger.info("msg.getTo() contains: " + msg.getTo().toString());
|
||||
logger.info("Trying to send message to " + recipient.toString());
|
||||
int port;
|
||||
try {
|
||||
port = domainLookup(recipient);
|
||||
logger.info("Domain lookup successful. Port is: " + port);
|
||||
if (!sent.get(recipient.getDomain())) {
|
||||
replayMessage(msg, port);
|
||||
sent.put(recipient.getDomain(), true);
|
||||
}
|
||||
} catch (UnknownDomainException e) {
|
||||
sendErrorMail(msg, e.getMessage());
|
||||
}
|
||||
}
|
||||
lock.notify();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void replayMessage(Message msg, int port) {
|
||||
logger.info("Replaying message for message: " + msg.toString() + " on port " + port);
|
||||
try {
|
||||
Socket socket = new Socket("127.0.0.1", port);
|
||||
PrintWriter socketOut = new PrintWriter(socket.getOutputStream(), true);
|
||||
BufferedReader socketIn = new BufferedReader(new InputStreamReader(socket.getInputStream()));
|
||||
socketIn.readLine();
|
||||
socketOut.println("begin");
|
||||
socketIn.readLine();
|
||||
socketOut.println("subject " + msg.getSubject());
|
||||
socketIn.readLine();
|
||||
socketOut.println("data " + msg.getData());
|
||||
socketIn.readLine();
|
||||
socketOut.println("to " + msg.printTo());
|
||||
socketIn.readLine();
|
||||
socketOut.println("from " + msg.getFrom().toString());
|
||||
socketIn.readLine();
|
||||
socketOut.println("send");
|
||||
String result = socketIn.readLine();
|
||||
if (result.startsWith("error"))
|
||||
sendErrorMail(msg, result);
|
||||
socketIn.close();
|
||||
socketOut.close();
|
||||
socket.close();
|
||||
sendMonitoringMessage(msg);
|
||||
} catch (IOException e) {
|
||||
sendErrorMail(msg, "error failed to connect to server");
|
||||
}
|
||||
}
|
||||
|
||||
private int domainLookup(Email email) throws UnknownDomainException {
|
||||
logger.info("Performing domain lookup for address " + email.toString());
|
||||
if (this.mailboxServers.containsKey(email.getDomain()))
|
||||
return this.mailboxServers.get(email.getDomain());
|
||||
throw new UnknownDomainException("error domain not found: " + email.getDomain());
|
||||
}
|
||||
|
||||
private void sendErrorMail(Message msg, String error) {
|
||||
logger.info("Trying to send error mail to address " + msg.getFrom());
|
||||
Message errorMessage = msg;
|
||||
int port;
|
||||
try {
|
||||
port = domainLookup(errorMessage.getFrom());
|
||||
} catch (UnknownDomainException e) {
|
||||
logger.severe("Sending error mail failed because sender domain is unknown");
|
||||
return;
|
||||
}
|
||||
ArrayList<Email> newTo = new ArrayList<>();
|
||||
newTo.add(errorMessage.getFrom());
|
||||
errorMessage.setTo(newTo);
|
||||
try {
|
||||
errorMessage.setFrom(new Email("mailer@127.0.0.1"));
|
||||
} catch (MalformedInputException e) {
|
||||
logger.severe("The server's E-Mail address is wrong. This should not be happening!");
|
||||
e.printStackTrace();
|
||||
return;
|
||||
}
|
||||
try {
|
||||
Socket socket = new Socket("127.0.0.1", port);
|
||||
PrintWriter socketOut = new PrintWriter(socket.getOutputStream(), true);
|
||||
BufferedReader socketIn = new BufferedReader(new InputStreamReader(socket.getInputStream()));
|
||||
socketIn.readLine();
|
||||
socketOut.println("begin");
|
||||
socketIn.readLine();
|
||||
socketOut.println("subject " + errorMessage.getSubject());
|
||||
socketIn.readLine();
|
||||
socketOut.println("data " + error);
|
||||
socketIn.readLine();
|
||||
socketOut.println("to " + errorMessage.printTo());
|
||||
socketIn.readLine();
|
||||
socketOut.println("from " + errorMessage.getFrom().toString());
|
||||
socketIn.readLine();
|
||||
socketOut.println("send");
|
||||
String result = socketIn.readLine();
|
||||
if (result.startsWith("error"))
|
||||
logger.severe("Sending error mail failed: " + result);
|
||||
socketIn.close();
|
||||
socketOut.close();
|
||||
socket.close();
|
||||
sendMonitoringMessage(errorMessage);
|
||||
} catch (IOException e) {
|
||||
logger.severe("Sending error mail failed because socket communication failed");
|
||||
}
|
||||
}
|
||||
|
||||
private void sendMonitoringMessage(Message msg) {
|
||||
DatagramSocket socket;
|
||||
try {
|
||||
socket = new DatagramSocket();
|
||||
} catch (SocketException e) {
|
||||
logger.severe("Failed to create DatagramSocket!");
|
||||
return;
|
||||
}
|
||||
byte[] sendBuffer = (transferHost + ":" + transferPort + " " + msg.getFrom().toString()).getBytes();
|
||||
try {
|
||||
DatagramPacket packet = new DatagramPacket(sendBuffer, sendBuffer.length, Inet4Address.getByName(monitoringHost), monitoringPort);
|
||||
socket.send(packet);
|
||||
} catch (IOException e) {
|
||||
logger.severe("Failed to send packet to " + monitoringHost + " on port " + monitoringPort);
|
||||
}
|
||||
socket.close();
|
||||
}
|
||||
|
||||
private void shutdown() {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,10 +1,8 @@
|
||||
# TODO: replace host and port with that of your server
|
||||
transfer.host=127.0.0.1
|
||||
transfer.port=port_range
|
||||
transfer.port=11620
|
||||
transfer.email=arthur@earth.planet
|
||||
|
||||
# TODO: replace host and port with that of your server
|
||||
mailbox.host=127.0.0.1
|
||||
mailbox.port=port_range+3
|
||||
mailbox.port=11623
|
||||
mailbox.user=arthur
|
||||
mailbox.password=23456
|
||||
|
||||
@@ -1,10 +1,8 @@
|
||||
# TODO: replace host and port with that of your server
|
||||
transfer.host=127.0.0.1
|
||||
transfer.port=port_range
|
||||
transfer.port=11620
|
||||
transfer.email=trillian@earth.planet
|
||||
|
||||
# TODO: replace host and port with that of your server
|
||||
mailbox.host=127.0.0.1
|
||||
mailbox.port=port_range+3
|
||||
mailbox.port=11623
|
||||
mailbox.user=trillian
|
||||
mailbox.password=12345
|
||||
|
||||
@@ -1,10 +1,8 @@
|
||||
# TODO: replace host and port with that of your server
|
||||
transfer.host=127.0.0.1
|
||||
transfer.port=port_range+1
|
||||
transfer.port=11621
|
||||
transfer.email=zaphod@univer.ze
|
||||
|
||||
# TODO: replace host and port with that of your server
|
||||
mailbox.host=127.0.0.1
|
||||
mailbox.port=port_range+5
|
||||
mailbox.port=11625
|
||||
mailbox.user=zaphod
|
||||
mailbox.password=12345
|
||||
|
||||
@@ -1,4 +1,3 @@
|
||||
# <domain>=<socket-address>
|
||||
# TODO: replace ports with the ports that your mailbox servers use
|
||||
earth.planet=127.0.0.1:port_range+2
|
||||
univer.ze=127.0.0.1:port_range+4
|
||||
earth.planet=127.0.0.1:11622
|
||||
univer.ze=127.0.0.1:11624
|
||||
|
||||
@@ -1,10 +1,8 @@
|
||||
# TCP port used for the DMTP server socket
|
||||
# TODO: REPLACE with real value such as 16502 - considering the port range associated with your account - you have received after Lab 0 a port range (beginning_of_the_range, end_of_the_range)
|
||||
dmtp.tcp.port=port_range+2
|
||||
dmtp.tcp.port=11622
|
||||
|
||||
# TCP port used for the DMAP server socket
|
||||
# TODO: REPLACE with real value such as 16503 - considering the port range associated with your account - you have received after Lab 0 a port range (beginning_of_the_range, end_of_the_range)
|
||||
dmap.tcp.port=port_range+3
|
||||
dmap.tcp.port=11623
|
||||
|
||||
# mail domain managed by the mailbox server
|
||||
domain=earth.planet
|
||||
@@ -21,5 +19,4 @@ root_id=root-nameserver
|
||||
registry.host=localhost
|
||||
|
||||
# RMI registry port
|
||||
# TODO: REPLACE with real value such as 16509 - considering the port range associated with your account - you have received after Lab 0 a port range (beginning_of_the_range, end_of_the_range)
|
||||
registry.port=port_range+9
|
||||
registry.port=11629
|
||||
|
||||
@@ -1,10 +1,8 @@
|
||||
# TCP port used for the DMTP server socket
|
||||
# TODO: REPLACE with real value such as 16504 - considering the port range associated with your account - you have received after Lab 0 a port range (beginning_of_the_range, end_of_the_range)
|
||||
dmtp.tcp.port=port_range+4
|
||||
dmtp.tcp.port=11624
|
||||
|
||||
# TCP port used for the DMAP server socket
|
||||
# TODO: REPLACE with real value such as 16505 - considering the port range associated with your account - you have received after Lab 0 a port range (beginning_of_the_range, end_of_the_range)
|
||||
dmap.tcp.port=port_range+5
|
||||
dmap.tcp.port=11625
|
||||
|
||||
# mail domain managed by the mailbox server
|
||||
domain=univer.ze
|
||||
@@ -21,5 +19,4 @@ root_id=root-nameserver
|
||||
registry.host=localhost
|
||||
|
||||
# RMI registry port
|
||||
# TODO: REPLACE with real value such as 16509 - considering the port range associated with your account - you have received after Lab 0 a port range (beginning_of_the_range, end_of_the_range)
|
||||
registry.port=port_range+9
|
||||
registry.port=11629
|
||||
|
||||
@@ -1,3 +1,2 @@
|
||||
# UDP port used for accepting monitoring packets
|
||||
# TODO: REPLACE with real value such as 16508 - considering the port range associated with your account - you have received after Lab 0 a port range (beginning_of_the_range, end_of_the_range)
|
||||
udp.port=port_range+8
|
||||
udp.port=11628
|
||||
|
||||
@@ -5,8 +5,7 @@ root_id=root-nameserver
|
||||
registry.host=localhost
|
||||
|
||||
# registry port
|
||||
# TODO: REPLACE with real value such as 16509 - considering the port range associated with your account - you have received after Lab 0 a port range (beginning_of_the_range, end_of_the_range)
|
||||
registry.port=port_range+9
|
||||
registry.port=11629
|
||||
|
||||
|
||||
# managed domain
|
||||
|
||||
@@ -5,8 +5,7 @@ root_id=root-nameserver
|
||||
registry.host=localhost
|
||||
|
||||
# registry port
|
||||
# TODO: REPLACE with real value such as 16509 - considering the port range associated with your account - you have received after Lab 0 a port range (beginning_of_the_range, end_of_the_range)
|
||||
registry.port=port_range+9
|
||||
registry.port=11629
|
||||
|
||||
|
||||
# managed domain
|
||||
|
||||
@@ -5,5 +5,4 @@ root_id=root-nameserver
|
||||
registry.host=localhost
|
||||
|
||||
# registry port
|
||||
# TODO: REPLACE with real value such as 16509 - considering the port range associated with your account - you have received after Lab 0 a port range (beginning_of_the_range, end_of_the_range)
|
||||
registry.port=port_range+9
|
||||
registry.port=11629
|
||||
|
||||
@@ -5,8 +5,7 @@ root_id=root-nameserver
|
||||
registry.host=localhost
|
||||
|
||||
# registry port
|
||||
# TODO: REPLACE with real value such as 16509 - considering the port range associated with your account - you have received after Lab 0 a port range (beginning_of_the_range, end_of_the_range)
|
||||
registry.port=port_range+9
|
||||
registry.port=11629
|
||||
|
||||
|
||||
# managed domain
|
||||
|
||||
@@ -1,11 +1,9 @@
|
||||
# TCP port used for the DMTP server socket
|
||||
# TODO: REPLACE with real value such as 16500 - considering the port range associated with your account - you have received after Lab 0 a port range (beginning_of_the_range, end_of_the_range)
|
||||
tcp.port=port_range
|
||||
tcp.port=11620
|
||||
|
||||
# UDP socket address of the monitoring server
|
||||
monitoring.host=127.0.0.1
|
||||
# TODO: REPLACE with the real value of the monitoring server port
|
||||
monitoring.port=port_range+8
|
||||
monitoring.port=11628
|
||||
|
||||
# ============================================= Required for Assignment 2
|
||||
|
||||
@@ -16,5 +14,4 @@ root_id=root-nameserver
|
||||
registry.host=localhost
|
||||
|
||||
# RMI registry port
|
||||
# TODO: REPLACE with real value such as 16509 - considering the port range associated with your account - you have received after Lab 0 a port range (beginning_of_the_range, end_of_the_range)
|
||||
registry.port=port_range+9
|
||||
registry.port=11629
|
||||
|
||||
@@ -1,11 +1,9 @@
|
||||
# TCP port used for the DMTP server socket
|
||||
# TODO: REPLACE with real value such as 16501 - considering the port range associated with your account - you have received after Lab 0 a port range (beginning_of_the_range, end_of_the_range)
|
||||
tcp.port=port_range+1
|
||||
tcp.port=11621
|
||||
|
||||
# UDP socket address of the monitoring server
|
||||
monitoring.host=127.0.0.1
|
||||
# TODO: REPLACE with the real value of the monitoring server port
|
||||
monitoring.port=port_range+8
|
||||
monitoring.port=11628
|
||||
|
||||
# ============================================= Required for Assignment 2
|
||||
|
||||
@@ -16,5 +14,4 @@ root_id=root-nameserver
|
||||
registry.host=localhost
|
||||
|
||||
# RMI registry port
|
||||
# TODO: REPLACE with real value such as 16509 - considering the port range associated with your account - you have received after Lab 0 a port range (beginning_of_the_range, end_of_the_range)
|
||||
registry.port=port_range+9
|
||||
registry.port=11629
|
||||
|
||||
Reference in New Issue
Block a user