diff --git a/L7_threads/src/main/java/ru/track/prefork/Client.java b/L7_threads/src/main/java/ru/track/prefork/Client.java index b0b61ee..b73c15d 100644 --- a/L7_threads/src/main/java/ru/track/prefork/Client.java +++ b/L7_threads/src/main/java/ru/track/prefork/Client.java @@ -1,5 +1,8 @@ package ru.track.prefork; +import java.io.InputStream; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; import java.io.OutputStream; import java.net.Socket; import java.util.Scanner; @@ -7,6 +10,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import static java.lang.System.*; + public class Client { static Logger log = LoggerFactory.getLogger(Client.class); @@ -21,15 +26,28 @@ public Client(int port, String host) { public void loop() throws Exception { Socket socket = new Socket(host, port); final OutputStream out = socket.getOutputStream(); + final InputStream in = socket.getInputStream(); + final ObjectOutputStream oos = new ObjectOutputStream(out); + final ObjectInputStream ois = new ObjectInputStream(in); Scanner scanner = new Scanner(System.in); - + Thread scannerThread; + scannerThread = new Thread(() -> { + try { + while (true) { + String line = scanner.nextLine(); + Message msg = new Message(line); + oos.writeObject(msg); + oos.flush(); + } + } catch (Exception e) { + e.printStackTrace(); + } + }); + scannerThread.setDaemon(true); + scannerThread.start(); while (true) { - String line = scanner.nextLine(); - if ("q".equals(line)) { - break; - } - out.write(line.getBytes()); - out.flush(); + Message msg = (Message) ois.readObject(); + System.out.println(msg.toString()); } } diff --git a/L7_threads/src/main/java/ru/track/prefork/Message.java b/L7_threads/src/main/java/ru/track/prefork/Message.java new file mode 100644 index 0000000..233161e --- /dev/null +++ b/L7_threads/src/main/java/ru/track/prefork/Message.java @@ -0,0 +1,22 @@ +package ru.track.prefork; + +import java.io.Serializable; + +public class Message implements Serializable { + public String username; + public String data; + + public Message(String username, String data) { + this.username = username; + this.data = data; + } + + public Message(String data) { + this.data = data; + } + + @Override + public String toString() { + return username + '>' + data; + } +} diff --git a/L7_threads/src/main/java/ru/track/prefork/Server.java b/L7_threads/src/main/java/ru/track/prefork/Server.java index bdeb794..f81e9eb 100644 --- a/L7_threads/src/main/java/ru/track/prefork/Server.java +++ b/L7_threads/src/main/java/ru/track/prefork/Server.java @@ -1,9 +1,13 @@ package ru.track.prefork; -import java.io.InputStream; +import java.io.*; import java.net.InetAddress; import java.net.ServerSocket; import java.net.Socket; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -19,22 +23,84 @@ */ public class Server { static Logger log = LoggerFactory.getLogger(Server.class); - + private AtomicInteger counter = new AtomicInteger(0); private int port; + Map threadMap = new HashMap<>(); + public Server(int port) { this.port = port; } + public class MyThread extends Thread { + private Socket socket; + private InputStream in; + private DataOutputStream out; + private ObjectOutputStream oos; + private int id; + + public MyThread(Socket socket, int id) throws IOException { + this.socket = socket; + setName(String.format("Client[%d]@%s:%d", id, socket.getInetAddress(), socket.getPort())); + this.id = id; + in = socket.getInputStream(); + out = new DataOutputStream(socket.getOutputStream()); + oos = new ObjectOutputStream(out); + } + + public void send(byte[] buf, int nRead, int id) { + try { + MyThread thread = threadMap.get(id); + StringBuilder sb = new StringBuilder(); + sb.append(String.format("Client[%d]@%s:%d>", id, thread.socket.getInetAddress(), thread.socket.getPort())).append(buf.toString()); + byte[] buff = sb.toString().getBytes(); + out.write(buff); + out.flush(); + } catch (IOException e) { + e.printStackTrace(); + } + } + + public void send(Message msg, int id) { + try { + MyThread thread = threadMap.get(id); + msg.username = String.format("Client[%d]@%s:%d", id, thread.socket.getInetAddress(), thread.socket.getPort()); + oos.writeObject(msg); + oos.flush(); + } catch (IOException e) { + e.printStackTrace(); + } + } + + public void run() { + try { + while (!socket.isClosed()) { + log.info("on select..."); + ObjectInputStream ois = new ObjectInputStream(in); + Message msg = (Message) ois.readObject(); + if ("EXIT".equals(msg.data)) { + socket.close(); + break; + } + for (Map.Entry entry : threadMap.entrySet()) { + if (entry.getValue().id != id) { + entry.getValue().send(msg, id); + } + } + } + } catch (Exception e) { + log.info("something is wrong"); + } + } + } + public void serve() throws Exception { ServerSocket serverSocket = new ServerSocket(port, 10, InetAddress.getByName("localhost")); while (true) { - log.info("on select..."); final Socket socket = serverSocket.accept(); - InputStream inputStream = socket.getInputStream(); - - byte[] buf = new byte[1024]; - int nRead = inputStream.read(buf); - System.out.println(new String(buf, 0, nRead)); + int id = counter.getAndIncrement(); + MyThread thread = new MyThread(socket, id); + threadMap.put(id, thread); + thread.start(); } } diff --git a/OWNER.md b/OWNER.md new file mode 100644 index 0000000..0e8f874 --- /dev/null +++ b/OWNER.md @@ -0,0 +1,2 @@ + +Name: Юнысов Илья