first commit
This commit is contained in:
@@ -0,0 +1,396 @@
|
||||
package org.distributed;
|
||||
import java.util.Map;
|
||||
import java.util.Scanner;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import javax.jms.*;
|
||||
import org.apache.activemq.ActiveMQConnectionFactory;
|
||||
|
||||
public class ChatHandler {
|
||||
|
||||
private String userId, corespondent;
|
||||
private Connection connection;
|
||||
private Session /*session,*/ senderSession, receiverSession;
|
||||
private int status = 0; // 0: offline, 1: online
|
||||
private final Map<String, MessageProducer> blocked = new ConcurrentHashMap<>();
|
||||
private final String TAG = "Task3_queue_";
|
||||
private final String BROKER_URL = "tcp://localhost:61616";
|
||||
/**
|
||||
* Initialize JMS session
|
||||
* @throws JMSException
|
||||
*/
|
||||
public void initializeSession() throws JMSException {
|
||||
ConnectionFactory factory = new ActiveMQConnectionFactory(BROKER_URL);
|
||||
this.connection = factory.createConnection();
|
||||
this.connection.start();
|
||||
//this.session = this.connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
|
||||
this.senderSession = this.connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
|
||||
this.receiverSession = this.connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
|
||||
}
|
||||
|
||||
/**
|
||||
* Start conversation loop
|
||||
* @param sc
|
||||
* @throws JMSException
|
||||
*/
|
||||
public void startConversation(Scanner sc) throws JMSException {
|
||||
|
||||
// init receiving thread
|
||||
Thread t = setupReceiver(userId);
|
||||
|
||||
// Continuous input loop
|
||||
try {
|
||||
while (sc.hasNextLine()){
|
||||
|
||||
// handle user input and send messages
|
||||
// or if exit command, break loop
|
||||
if(!respond(sc)){
|
||||
break;
|
||||
}
|
||||
}
|
||||
} catch (Exception e) {
|
||||
System.err.println("Error in continuous conversation: " + e.getMessage());
|
||||
} finally {
|
||||
t.interrupt();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle user input and send messages
|
||||
* @param sc
|
||||
* @throws JMSException
|
||||
*/
|
||||
private boolean respond(Scanner sc) throws JMSException {
|
||||
String line = sc.nextLine();
|
||||
|
||||
// commands to block or unblock contact
|
||||
if(line.startsWith("-")){
|
||||
delegateBlockingManagement(sc,line);
|
||||
//return true;
|
||||
}
|
||||
|
||||
if (line.equalsIgnoreCase("exit")) {
|
||||
System.out.println("Exiting conversation...");
|
||||
//return false;
|
||||
}
|
||||
|
||||
// Send message to receiver
|
||||
sendMessage(createProducer(corespondent), line);
|
||||
return !line.equalsIgnoreCase("exit");//true;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Delegate blocking and unblocking management
|
||||
* @param sc
|
||||
* @param line
|
||||
*/
|
||||
private void delegateBlockingManagement(Scanner sc, String line) {
|
||||
|
||||
// block command is -b followed by contact id
|
||||
if(line.startsWith("-b")){
|
||||
blockContact(line);
|
||||
promptForCorespondent(sc);
|
||||
}
|
||||
|
||||
// unblock command is -u followed by contact id
|
||||
if(line.startsWith("-u")){
|
||||
String contact = unBlockContact(line);
|
||||
System.out.println("do you want to chat with: " + contact+" yes or no");
|
||||
while(true){
|
||||
String answer = sc.nextLine();
|
||||
if(answer.equalsIgnoreCase("yes")){
|
||||
setCorespondent(contact);
|
||||
break;
|
||||
} else if(answer.equalsIgnoreCase("no")){
|
||||
promptForCorespondent(sc);
|
||||
break;
|
||||
} else {
|
||||
System.out.println("please answer with yes or no");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Setup producer and consumer for a receiver
|
||||
private Thread setupReceiver(String receiver) throws JMSException {
|
||||
MessageConsumer contact = createConsumer();
|
||||
|
||||
// add consumer to contacts map
|
||||
//contacts.put(receiver, contact);
|
||||
return getThread(contact);
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a producer for a given contact ID so we can send messages
|
||||
* @param id
|
||||
* @return MessageProducer
|
||||
* @throws JMSException
|
||||
*/
|
||||
private MessageProducer createProducer(String id) throws JMSException {
|
||||
String from = TAG + id;
|
||||
Queue prod = senderSession.createQueue(from);
|
||||
return senderSession.createProducer(prod);
|
||||
}
|
||||
|
||||
/**
|
||||
* Block a contact
|
||||
* @param command
|
||||
*/
|
||||
private void blockContact(String command) {
|
||||
// remove the command part to get the contact id
|
||||
String contact = command.replace("-b", "").trim();
|
||||
try {
|
||||
MessageProducer ignored = createProducer(contact);
|
||||
// remove from contacts map
|
||||
// and close his ass
|
||||
if(ignored != null){
|
||||
ignored.close();
|
||||
blocked.put(contact, ignored);
|
||||
System.out.println(contact + " has been blocked.");
|
||||
}
|
||||
} catch (JMSException e) {
|
||||
System.err.println("Error unblocking consumer: " + e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Unblock a contact
|
||||
* @param command
|
||||
* @return contact id
|
||||
*/
|
||||
private String unBlockContact(String command) {
|
||||
|
||||
String contact = command.replace("-u", "").trim();
|
||||
|
||||
|
||||
try {
|
||||
// unblock by removing his ass from blocklist
|
||||
blocked.remove(contact);
|
||||
// retun contact name for further handling
|
||||
return contact;
|
||||
|
||||
} catch (Exception e) {
|
||||
System.err.println("Error blocking consumer: " + e.getMessage());
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Subscribe to topics meant for this user
|
||||
* @return MessageConsumer
|
||||
* @throws JMSException
|
||||
*/
|
||||
private MessageConsumer createConsumer() throws JMSException {
|
||||
String to = TAG + userId;
|
||||
Queue cons = receiverSession.createQueue(to);
|
||||
return receiverSession.createConsumer(cons);
|
||||
}
|
||||
|
||||
/**
|
||||
* Send a messag
|
||||
* @param producer
|
||||
* @param line
|
||||
* @throws JMSException
|
||||
*/
|
||||
private void sendMessage(MessageProducer producer, String line) throws JMSException {
|
||||
|
||||
TextMessage m = senderSession.createTextMessage(line);
|
||||
// add sender identity to message
|
||||
m.setStringProperty("sender", userId);
|
||||
producer.send(m);
|
||||
}
|
||||
|
||||
/**
|
||||
* Close session and connection
|
||||
* @throws JMSException
|
||||
*/
|
||||
public void closeSession() throws JMSException {
|
||||
if (senderSession != null) {
|
||||
senderSession.close();
|
||||
}
|
||||
if( receiverSession != null) {
|
||||
receiverSession.close();
|
||||
}
|
||||
if (connection != null) {
|
||||
connection.close();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns a thread object to handle receiving messages
|
||||
* @param consumer
|
||||
* @return Thread
|
||||
*/
|
||||
private Thread getThread(MessageConsumer consumer) {
|
||||
|
||||
Thread receivingThread = new Thread(() -> {
|
||||
try {
|
||||
|
||||
while (!Thread.currentThread().isInterrupted()) {
|
||||
|
||||
// receive message with timeout
|
||||
Message msg = consumer.receive(100); // 100ms timeout
|
||||
if (msg instanceof TextMessage tm) {
|
||||
// handle received message
|
||||
String sender = tm.getStringProperty("sender");
|
||||
|
||||
// display message if sender is not blocked
|
||||
if (!isBlocked(sender)) {
|
||||
String body = tm.getText();
|
||||
displayMessage(sender, body);
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
} catch (javax.jms.JMSException e) {
|
||||
// Silently handle interruption - normal shutdown behavior
|
||||
if (!Thread.currentThread().isInterrupted()) {
|
||||
System.err.println("Error receiving message: " + e.getMessage());
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
|
||||
receivingThread.start();
|
||||
// we are returning this so we can interrupt it later
|
||||
return receivingThread;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Display received message
|
||||
* Checks if sender is blocked
|
||||
* checks if message is empty for status update
|
||||
* checks for exit condition
|
||||
* @param tm
|
||||
* @throws JMSException
|
||||
*/
|
||||
|
||||
private void displayMessage(String sender, String body) {
|
||||
|
||||
|
||||
// Exit condition because correspondent went offline
|
||||
if(body.equalsIgnoreCase("exit")) {
|
||||
setStatus(0);
|
||||
return;
|
||||
}
|
||||
|
||||
// display only if not empty
|
||||
if(!body.isBlank() || !body.isEmpty()) {
|
||||
// Status handling for online/offline
|
||||
System.out.println("[" + sender + "]: " + body);
|
||||
}
|
||||
setStatus(1);
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if sender is blocked
|
||||
* if sender is blocked, print info message
|
||||
* @param sender
|
||||
* @return
|
||||
*/
|
||||
private boolean isBlocked(String sender) {
|
||||
boolean isBlocked = blocked.containsKey(sender);
|
||||
if(isBlocked){
|
||||
System.out.println(" >> Message from '"+sender+"' ignored. because user is blocked.");
|
||||
}
|
||||
return isBlocked;
|
||||
}
|
||||
|
||||
/**
|
||||
* Prompt for user ID
|
||||
* @param sc
|
||||
*/
|
||||
public void promptForUser(Scanner sc){
|
||||
|
||||
System.out.print("Enter your user ID: ");
|
||||
//userId = prompt(sc);
|
||||
setUserId(prompt(sc));
|
||||
// Instructions
|
||||
System.out.print("Welcome "+getUserId());
|
||||
System.out.println(" -> Type 'exit' to quit.");
|
||||
}
|
||||
|
||||
/**
|
||||
* Prompt for correspondent ID
|
||||
* @param sc
|
||||
*/
|
||||
public void promptForCorespondent(Scanner sc){
|
||||
|
||||
System.out.print("Who do you want to chat with? ");
|
||||
String contact = prompt(sc);
|
||||
setCorespondent(contact);
|
||||
// check if contact is blocked
|
||||
if(blocked.containsKey(contact)){
|
||||
System.out.println(contact + " is currently blocked. Unblock to chat.");
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Generic prompt method prompts for whatever we need
|
||||
* @param sc Scanner object
|
||||
* @return String
|
||||
*/
|
||||
private String prompt(Scanner sc) {
|
||||
|
||||
String user = "";
|
||||
|
||||
while (user.isEmpty()) {
|
||||
try {
|
||||
user = sc.next();
|
||||
|
||||
} catch (Exception e) {
|
||||
System.err.println("Error reading correspondent ID: " + e.getMessage());
|
||||
return null;
|
||||
}
|
||||
}
|
||||
return user;
|
||||
}
|
||||
|
||||
// Setters and getters
|
||||
// user
|
||||
public void setUserId(String userId) {
|
||||
this.userId = userId;
|
||||
}
|
||||
|
||||
public String getUserId() {
|
||||
return userId;
|
||||
}
|
||||
|
||||
// correspondent
|
||||
public String getCorespondent() {
|
||||
return corespondent;
|
||||
}
|
||||
|
||||
public void setCorespondent(String corespondent) {
|
||||
this.corespondent = corespondent;
|
||||
System.out.println("You can chat with: " + corespondent);
|
||||
// reset old status when new corepondent is set
|
||||
setStatus(0);
|
||||
}
|
||||
|
||||
/**
|
||||
* Updates corespondents status
|
||||
* @param status
|
||||
*/
|
||||
private void setStatus(int status) {
|
||||
|
||||
// just in case of duplicate status messages, if status is same, return
|
||||
if(this.status == status) {
|
||||
return;
|
||||
}
|
||||
|
||||
// update status
|
||||
this.status = status;
|
||||
|
||||
// and print accordingly
|
||||
if(status == 0) {
|
||||
System.out.println(getCorespondent()+" is now offline.");
|
||||
} else {
|
||||
System.out.println(getCorespondent()+" is now online.");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
package org.distributed;
|
||||
import java.util.Scanner;
|
||||
public class Main {
|
||||
|
||||
// entry point
|
||||
public static void main(String[] args) throws Exception {
|
||||
|
||||
// create a client
|
||||
ChatHandler handler = new ChatHandler();
|
||||
|
||||
// init session
|
||||
handler.initializeSession();
|
||||
|
||||
// create scanner for user input
|
||||
Scanner sc = new Scanner(System.in);
|
||||
|
||||
// log in user
|
||||
handler.promptForUser(sc);
|
||||
|
||||
// specify correspondent
|
||||
handler.promptForCorespondent(sc);
|
||||
|
||||
// start conversation
|
||||
handler.startConversation(sc);
|
||||
|
||||
// close session after all receivers are processed
|
||||
handler.closeSession();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user