From 5e18243281330e95301e2afa7119d8665223a42c Mon Sep 17 00:00:00 2001 From: Highoc Date: Wed, 21 Feb 2018 19:54:29 +0300 Subject: [PATCH 1/6] Added OWNER --- OWNER.md | 1 + 1 file changed, 1 insertion(+) create mode 100644 OWNER.md diff --git a/OWNER.md b/OWNER.md new file mode 100644 index 0000000..8b2f6cb --- /dev/null +++ b/OWNER.md @@ -0,0 +1 @@ +Морев Максим Сергеевич From 31c41919ec1742d850207ce7dfeda1d91e356b12 Mon Sep 17 00:00:00 2001 From: Highoc Date: Sun, 22 Apr 2018 19:57:37 +0300 Subject: [PATCH 2/6] 1-2 --- .gitignore | 1 + .../main/java/ru/track/homework/Client.java | 98 +++++++++++++++++++ .../java/ru/track/homework/HandlerThread.java | 52 ++++++++++ .../main/java/ru/track/homework/Message.java | 12 +++ .../main/java/ru/track/homework/Server.java | 76 ++++++++++++++ 5 files changed, 239 insertions(+) create mode 100644 L7_threads/src/main/java/ru/track/homework/Client.java create mode 100644 L7_threads/src/main/java/ru/track/homework/HandlerThread.java create mode 100644 L7_threads/src/main/java/ru/track/homework/Message.java create mode 100644 L7_threads/src/main/java/ru/track/homework/Server.java diff --git a/.gitignore b/.gitignore index e32026c..1ee3778 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,4 @@ **/*.iml /.idea **/target/ +out/ diff --git a/L7_threads/src/main/java/ru/track/homework/Client.java b/L7_threads/src/main/java/ru/track/homework/Client.java new file mode 100644 index 0000000..d9141bc --- /dev/null +++ b/L7_threads/src/main/java/ru/track/homework/Client.java @@ -0,0 +1,98 @@ +package ru.track.homework; + +import org.apache.commons.io.IOUtils; +import org.jetbrains.annotations.NotNull; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.Socket; +import java.util.Scanner; + +public class Client { + + private final int port; + private final String host; + + private Thread writer; + private Thread reader; + + public Client(@NotNull String host, int port) { + this.port = port; + this.host = host; + + writer = null; + reader = null; + } + + private void connect() + { + Socket socket = null; + try { + socket = new Socket(host, port); + final Socket finalSocket = socket; + + writer = new Thread(() -> { System.out.println("writer > started"); writeToSocket(finalSocket); }); + writer.setName("writer"); + writer.start(); + + reader = new Thread(() -> { System.out.println("reader > started"); readFromSocket(finalSocket); }); + reader.setName("reader"); + reader.start(); + + } catch (IOException e) { + throw new RuntimeException(e); + } + } + + private void readFromSocket(@NotNull Socket socket) { + try { + final InputStream in = socket.getInputStream(); + final byte[] buffer = new byte[4096]; + + for (int nRead = 0; nRead != -1; nRead = in.read(buffer)) { + if (nRead == 0) continue; + + System.out.println(new String(buffer, 0, nRead)); + } + socket.shutdownInput(); + + } catch (IOException e) { + throw new RuntimeException(e); + } finally { + IOUtils.closeQuietly(socket); + } + + } + private void writeToSocket(@NotNull Socket socket) { + try{ + final OutputStream out = socket.getOutputStream(); + Scanner scanner = new Scanner(System.in); + + while (true) { + String line = scanner.nextLine(); + if ("exit".equals(line)) { + System.out.println("bye!"); + break; + } + + out.write(line.getBytes()); + out.flush(); + } + socket.shutdownOutput(); + + } catch (IOException e) { + throw new RuntimeException(e); + } finally { + IOUtils.closeQuietly(socket); + } + } + + + public static void main(String[] args) throws Exception { + Client client = new Client("localhost",9000); + client.connect(); + } +} diff --git a/L7_threads/src/main/java/ru/track/homework/HandlerThread.java b/L7_threads/src/main/java/ru/track/homework/HandlerThread.java new file mode 100644 index 0000000..c297415 --- /dev/null +++ b/L7_threads/src/main/java/ru/track/homework/HandlerThread.java @@ -0,0 +1,52 @@ +package ru.track.homework; + +import org.apache.commons.io.IOUtils; +import org.jetbrains.annotations.NotNull; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.Socket; + +public class HandlerThread extends Thread { + + private Socket socket; + private int clientID; + + private static int counterID = 0; + + HandlerThread(@NotNull Socket socket) { + this.socket = socket; + clientID = counterID++; + + setName(String.format("Client[%d]@%s:%s", clientID, socket.getInetAddress(), socket.getPort())); + } + + public void run() { + final byte[] buffer = new byte[4096]; + + try + { + final InputStream in = socket.getInputStream(); + final OutputStream out = socket.getOutputStream(); + + for (int nRead = 0; nRead != -1; nRead = in.read(buffer)) { + if (nRead == 0) continue; + System.out.println(String.format("\t%s > %s", getName(), new String(buffer, 0, nRead))); + out.write(buffer, 0, nRead); + out.flush(); + } + + socket.shutdownInput(); + socket.shutdownOutput(); + + } catch (IOException e) { + throw new RuntimeException(e); + } finally { + System.out.println(String.format("main > client disconnected: %s", getName())); + IOUtils.closeQuietly(socket); + } + } +} diff --git a/L7_threads/src/main/java/ru/track/homework/Message.java b/L7_threads/src/main/java/ru/track/homework/Message.java new file mode 100644 index 0000000..ec0d537 --- /dev/null +++ b/L7_threads/src/main/java/ru/track/homework/Message.java @@ -0,0 +1,12 @@ +package ru.track.homework; + +class Message { + private long ts; + private String data; + + public Message(long ts, String data) + { + this.ts = ts; + this.data = data; + } +} \ No newline at end of file diff --git a/L7_threads/src/main/java/ru/track/homework/Server.java b/L7_threads/src/main/java/ru/track/homework/Server.java new file mode 100644 index 0000000..80ddeb1 --- /dev/null +++ b/L7_threads/src/main/java/ru/track/homework/Server.java @@ -0,0 +1,76 @@ +package ru.track.homework; + +import org.apache.commons.io.IOUtils; +import org.jetbrains.annotations.NotNull; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.util.*; + + +public class Server { + + private final int port; + private final int backlog; + + //private Map clients = new HashMap<>(); + private List clientThreads = new ArrayList<>(); + private static int counterID = 0; + + class Client { + private int clientID; + private Socket socket; + + private Thread reader; + private Thread writer; + + private Queue messageQueue; + } + + public Server(int port, int backlog) { + this.port = port; + this.backlog = backlog; + } + + public void serve() throws IOException { + ServerSocket serverSocket = null; + try { + final ServerSocket serverSocketFinal = new ServerSocket(port, backlog, InetAddress.getByName("localhost")); + serverSocket = serverSocketFinal; + + while (true){ + System.out.println("main > wait to accept new client..."); + handle(serverSocketFinal); + } + + } finally { + IOUtils.closeQuietly(serverSocket); + } + } + + private void handle(@NotNull ServerSocket serverSocket) { + Socket socket = null; + try { + socket = serverSocket.accept(); + + Thread thread = new HandlerThread(socket); + clientThreads.add(thread); + System.out.println(String.format("main > new client connected: %s", thread.getName())); + thread.start(); + + } catch (IOException e) { + throw new RuntimeException(e); + } + } + + public static void main(String[] args) throws Exception { + Server server = new Server(9000, 10); + server.serve(); + } +} From 7e4fbcb589f4d87e8b5ed32f3a09eb5632485610 Mon Sep 17 00:00:00 2001 From: Highoc Date: Wed, 25 Apr 2018 10:28:00 +0300 Subject: [PATCH 3/6] v-1.0 --- .../main/java/ru/track/homework/Client.java | 8 +- .../java/ru/track/homework/ClientThread.java | 73 +++++++++++++++++++ .../java/ru/track/homework/HandlerThread.java | 52 ------------- .../main/java/ru/track/homework/Message.java | 12 ++- .../main/java/ru/track/homework/Server.java | 58 ++++++++++----- 5 files changed, 123 insertions(+), 80 deletions(-) create mode 100644 L7_threads/src/main/java/ru/track/homework/ClientThread.java delete mode 100644 L7_threads/src/main/java/ru/track/homework/HandlerThread.java diff --git a/L7_threads/src/main/java/ru/track/homework/Client.java b/L7_threads/src/main/java/ru/track/homework/Client.java index d9141bc..a7dc0e2 100644 --- a/L7_threads/src/main/java/ru/track/homework/Client.java +++ b/L7_threads/src/main/java/ru/track/homework/Client.java @@ -22,16 +22,12 @@ public class Client { public Client(@NotNull String host, int port) { this.port = port; this.host = host; - - writer = null; - reader = null; } private void connect() { - Socket socket = null; try { - socket = new Socket(host, port); + Socket socket = new Socket(host, port); final Socket finalSocket = socket; writer = new Thread(() -> { System.out.println("writer > started"); writeToSocket(finalSocket); }); @@ -64,8 +60,8 @@ private void readFromSocket(@NotNull Socket socket) { } finally { IOUtils.closeQuietly(socket); } - } + private void writeToSocket(@NotNull Socket socket) { try{ final OutputStream out = socket.getOutputStream(); diff --git a/L7_threads/src/main/java/ru/track/homework/ClientThread.java b/L7_threads/src/main/java/ru/track/homework/ClientThread.java new file mode 100644 index 0000000..8e5e821 --- /dev/null +++ b/L7_threads/src/main/java/ru/track/homework/ClientThread.java @@ -0,0 +1,73 @@ +package ru.track.homework; + +import org.apache.commons.io.IOUtils; +import org.jetbrains.annotations.NotNull; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.util.Date; +import java.util.List; +import java.util.Queue; + +public class ClientThread extends Thread { + + private final int clientID; + private static int counterID = 0; + + private static List messages; + private final Socket socket; + + private InputStream in; + private OutputStream out; + + public ClientThread(@NotNull Socket socket, @NotNull List messages) { + this.clientID = counterID++; + this.socket = socket; + this.messages = messages; + setName(String.format("Client[%d]@%s:%s", clientID, socket.getInetAddress(), socket.getPort())); + } + + public void run() + { + try { + in = socket.getInputStream(); + out = socket.getOutputStream(); + + final byte[] buffer = new byte[4096]; + + for (int nRead = 0; nRead != -1; nRead = in.read(buffer)) { + if (nRead == 0) continue; + + Date date = new Date(); + Message message = new Message(getName(), new String(buffer, 0, nRead), date.getTime()); + messages.add(message); + System.out.println(String.format("%s > %s", message.getSenderName(), message.getData())); + } + + socket.shutdownInput(); + socket.shutdownOutput(); + + } catch (IOException e) { + throw new RuntimeException(e); + } finally { + System.out.println(String.format("main > client disconnected: %s", getName())); + IOUtils.closeQuietly(socket); + } + } + + public void send(Message message) + { + try { + final byte[] buffer = message.getData().getBytes(); + out.write(buffer); + out.flush(); + + } catch (IOException e) { + throw new RuntimeException(e); + } + } +} diff --git a/L7_threads/src/main/java/ru/track/homework/HandlerThread.java b/L7_threads/src/main/java/ru/track/homework/HandlerThread.java deleted file mode 100644 index c297415..0000000 --- a/L7_threads/src/main/java/ru/track/homework/HandlerThread.java +++ /dev/null @@ -1,52 +0,0 @@ -package ru.track.homework; - -import org.apache.commons.io.IOUtils; -import org.jetbrains.annotations.NotNull; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.IOException; -import java.io.InputStream; -import java.io.OutputStream; -import java.net.Socket; - -public class HandlerThread extends Thread { - - private Socket socket; - private int clientID; - - private static int counterID = 0; - - HandlerThread(@NotNull Socket socket) { - this.socket = socket; - clientID = counterID++; - - setName(String.format("Client[%d]@%s:%s", clientID, socket.getInetAddress(), socket.getPort())); - } - - public void run() { - final byte[] buffer = new byte[4096]; - - try - { - final InputStream in = socket.getInputStream(); - final OutputStream out = socket.getOutputStream(); - - for (int nRead = 0; nRead != -1; nRead = in.read(buffer)) { - if (nRead == 0) continue; - System.out.println(String.format("\t%s > %s", getName(), new String(buffer, 0, nRead))); - out.write(buffer, 0, nRead); - out.flush(); - } - - socket.shutdownInput(); - socket.shutdownOutput(); - - } catch (IOException e) { - throw new RuntimeException(e); - } finally { - System.out.println(String.format("main > client disconnected: %s", getName())); - IOUtils.closeQuietly(socket); - } - } -} diff --git a/L7_threads/src/main/java/ru/track/homework/Message.java b/L7_threads/src/main/java/ru/track/homework/Message.java index ec0d537..84743a4 100644 --- a/L7_threads/src/main/java/ru/track/homework/Message.java +++ b/L7_threads/src/main/java/ru/track/homework/Message.java @@ -1,12 +1,20 @@ package ru.track.homework; +import org.jetbrains.annotations.NotNull; + class Message { - private long ts; + private String senderName; private String data; + private long ts; - public Message(long ts, String data) + public Message(@NotNull String senderName, @NotNull String data, long ts) { + this.senderName = senderName; this.ts = ts; this.data = data; } + + public String getData() { return data; } + public String getSenderName() { return senderName; } + public long getTimeStand() { return ts; } } \ No newline at end of file diff --git a/L7_threads/src/main/java/ru/track/homework/Server.java b/L7_threads/src/main/java/ru/track/homework/Server.java index 80ddeb1..bbce954 100644 --- a/L7_threads/src/main/java/ru/track/homework/Server.java +++ b/L7_threads/src/main/java/ru/track/homework/Server.java @@ -15,27 +15,18 @@ public class Server { - private final int port; private final int backlog; - //private Map clients = new HashMap<>(); - private List clientThreads = new ArrayList<>(); - private static int counterID = 0; - - class Client { - private int clientID; - private Socket socket; - - private Thread reader; - private Thread writer; - - private Queue messageQueue; - } + private List messages = Collections.synchronizedList(new ArrayList<>()); + private List clients = new ArrayList<>(); + private Thread broadcaster = new Thread(() -> { broadcast(); });; public Server(int port, int backlog) { this.port = port; this.backlog = backlog; + + broadcaster.start(); } public void serve() throws IOException { @@ -55,20 +46,47 @@ public void serve() throws IOException { } private void handle(@NotNull ServerSocket serverSocket) { - Socket socket = null; try { - socket = serverSocket.accept(); + Socket socket = serverSocket.accept(); - Thread thread = new HandlerThread(socket); - clientThreads.add(thread); - System.out.println(String.format("main > new client connected: %s", thread.getName())); - thread.start(); + ClientThread client = new ClientThread(socket, messages); + clients.add(client); + client.start(); + System.out.println(String.format("main > new client connected: %s", client.getName())); } catch (IOException e) { throw new RuntimeException(e); } } + private void broadcast() + { + try { + Date date = new Date(); + long lastUpdateTime = date.getTime(); + + while (true) { + synchronized (messages) { + for (Message messsage: messages) { + if (messsage.getTimeStand() > lastUpdateTime) + { + for (ClientThread client: clients) { + if (!client.getName().equals(messsage.getSenderName())) + { + client.send(messsage); + } + } + } + } + lastUpdateTime = date.getTime(); + } + Thread.sleep(2000); + } + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } + public static void main(String[] args) throws Exception { Server server = new Server(9000, 10); server.serve(); From 53c9d959d8ee2b8ddbfd24d2b554274a30094766 Mon Sep 17 00:00:00 2001 From: Highoc Date: Wed, 25 Apr 2018 17:24:50 +0300 Subject: [PATCH 4/6] d --- L7_threads/pom.xml | 7 +++ .../main/java/ru/track/homework/Client.java | 20 +++--- .../java/ru/track/homework/ClientThread.java | 16 ++--- .../main/java/ru/track/homework/Message.java | 1 + .../main/java/ru/track/homework/Server.java | 62 ++++++++++++++++--- 5 files changed, 84 insertions(+), 22 deletions(-) diff --git a/L7_threads/pom.xml b/L7_threads/pom.xml index c0d9e38..2bb00c4 100644 --- a/L7_threads/pom.xml +++ b/L7_threads/pom.xml @@ -94,5 +94,12 @@ 1.7.5 + + + com.google.code.gson + gson + 2.8.2 + + \ No newline at end of file diff --git a/L7_threads/src/main/java/ru/track/homework/Client.java b/L7_threads/src/main/java/ru/track/homework/Client.java index a7dc0e2..bc09bd4 100644 --- a/L7_threads/src/main/java/ru/track/homework/Client.java +++ b/L7_threads/src/main/java/ru/track/homework/Client.java @@ -1,14 +1,14 @@ package ru.track.homework; +import com.google.gson.Gson; import org.apache.commons.io.IOUtils; import org.jetbrains.annotations.NotNull; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; import java.net.Socket; +import java.util.Date; import java.util.Scanner; public class Client { @@ -19,6 +19,8 @@ public class Client { private Thread writer; private Thread reader; + private static Gson gson = new Gson(); + public Client(@NotNull String host, int port) { this.port = port; this.host = host; @@ -30,11 +32,11 @@ private void connect() Socket socket = new Socket(host, port); final Socket finalSocket = socket; - writer = new Thread(() -> { System.out.println("writer > started"); writeToSocket(finalSocket); }); + writer = new Thread(() -> { writeToSocket(finalSocket); }); writer.setName("writer"); writer.start(); - reader = new Thread(() -> { System.out.println("reader > started"); readFromSocket(finalSocket); }); + reader = new Thread(() -> { readFromSocket(finalSocket); }); reader.setName("reader"); reader.start(); @@ -51,7 +53,8 @@ private void readFromSocket(@NotNull Socket socket) { for (int nRead = 0; nRead != -1; nRead = in.read(buffer)) { if (nRead == 0) continue; - System.out.println(new String(buffer, 0, nRead)); + Message message = gson.fromJson(new String(buffer, 0, nRead), Message.class); + System.out.println(String.format("%s> %s", message.getSenderName(), message.getData())); } socket.shutdownInput(); @@ -74,7 +77,11 @@ private void writeToSocket(@NotNull Socket socket) { break; } - out.write(line.getBytes()); + Date date = new Date(); + Message message = new Message("", line, date.getTime()); + String result = gson.toJson(message); + + out.write(result.getBytes()); out.flush(); } socket.shutdownOutput(); @@ -86,7 +93,6 @@ private void writeToSocket(@NotNull Socket socket) { } } - public static void main(String[] args) throws Exception { Client client = new Client("localhost",9000); client.connect(); diff --git a/L7_threads/src/main/java/ru/track/homework/ClientThread.java b/L7_threads/src/main/java/ru/track/homework/ClientThread.java index 8e5e821..6fc9130 100644 --- a/L7_threads/src/main/java/ru/track/homework/ClientThread.java +++ b/L7_threads/src/main/java/ru/track/homework/ClientThread.java @@ -1,5 +1,6 @@ package ru.track.homework; +import com.google.gson.Gson; import org.apache.commons.io.IOUtils; import org.jetbrains.annotations.NotNull; @@ -24,6 +25,8 @@ public class ClientThread extends Thread { private InputStream in; private OutputStream out; + private static Gson gson = new Gson(); + public ClientThread(@NotNull Socket socket, @NotNull List messages) { this.clientID = counterID++; this.socket = socket; @@ -42,10 +45,10 @@ public void run() for (int nRead = 0; nRead != -1; nRead = in.read(buffer)) { if (nRead == 0) continue; - Date date = new Date(); - Message message = new Message(getName(), new String(buffer, 0, nRead), date.getTime()); + Message message = gson.fromJson(new String(buffer, 0, nRead), Message.class); + message.setSenderName(getName()); messages.add(message); - System.out.println(String.format("%s > %s", message.getSenderName(), message.getData())); + System.out.println(String.format("%s> %s", message.getSenderName(), message.getData())); } socket.shutdownInput(); @@ -54,7 +57,7 @@ public void run() } catch (IOException e) { throw new RuntimeException(e); } finally { - System.out.println(String.format("main > client disconnected: %s", getName())); + System.out.println(String.format("main> client disconnected: %s", getName())); IOUtils.closeQuietly(socket); } } @@ -62,10 +65,9 @@ public void run() public void send(Message message) { try { - final byte[] buffer = message.getData().getBytes(); - out.write(buffer); + String result = gson.toJson(message); + out.write(result.getBytes()); out.flush(); - } catch (IOException e) { throw new RuntimeException(e); } diff --git a/L7_threads/src/main/java/ru/track/homework/Message.java b/L7_threads/src/main/java/ru/track/homework/Message.java index 84743a4..6c81285 100644 --- a/L7_threads/src/main/java/ru/track/homework/Message.java +++ b/L7_threads/src/main/java/ru/track/homework/Message.java @@ -16,5 +16,6 @@ public Message(@NotNull String senderName, @NotNull String data, long ts) public String getData() { return data; } public String getSenderName() { return senderName; } + public void setSenderName(@NotNull String senderName) { this.senderName = senderName; } public long getTimeStand() { return ts; } } \ No newline at end of file diff --git a/L7_threads/src/main/java/ru/track/homework/Server.java b/L7_threads/src/main/java/ru/track/homework/Server.java index bbce954..ac2bd92 100644 --- a/L7_threads/src/main/java/ru/track/homework/Server.java +++ b/L7_threads/src/main/java/ru/track/homework/Server.java @@ -5,6 +5,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.io.BufferedInputStream; import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; @@ -13,20 +14,22 @@ import java.net.Socket; import java.util.*; +//Не пробрасывать id на клиент +//Корректная работа со списком клиентов +//Админская консоль public class Server { private final int port; private final int backlog; private List messages = Collections.synchronizedList(new ArrayList<>()); - private List clients = new ArrayList<>(); - private Thread broadcaster = new Thread(() -> { broadcast(); });; + private List clients = Collections.synchronizedList(new ArrayList<>()); + private Thread broadcasterThread = new Thread(() -> { broadcast(); }); + private Thread adminConsoleThread = new Thread(() -> { adminConsole(); }); public Server(int port, int backlog) { this.port = port; this.backlog = backlog; - - broadcaster.start(); } public void serve() throws IOException { @@ -35,8 +38,11 @@ public void serve() throws IOException { final ServerSocket serverSocketFinal = new ServerSocket(port, backlog, InetAddress.getByName("localhost")); serverSocket = serverSocketFinal; + broadcasterThread.start(); + adminConsoleThread.start(); + while (true){ - System.out.println("main > wait to accept new client..."); + System.out.println("main> wait to accept new client..."); handle(serverSocketFinal); } @@ -53,7 +59,7 @@ private void handle(@NotNull ServerSocket serverSocket) { clients.add(client); client.start(); - System.out.println(String.format("main > new client connected: %s", client.getName())); + System.out.println(String.format("main> new client connected: %s", client.getName())); } catch (IOException e) { throw new RuntimeException(e); } @@ -61,6 +67,7 @@ private void handle(@NotNull ServerSocket serverSocket) { private void broadcast() { + BufferedInputStream try { Date date = new Date(); long lastUpdateTime = date.getTime(); @@ -68,9 +75,13 @@ private void broadcast() while (true) { synchronized (messages) { for (Message messsage: messages) { - if (messsage.getTimeStand() > lastUpdateTime) + if (messsage.getTimeStand() >= lastUpdateTime) { for (ClientThread client: clients) { + if (!client.isAlive()) { + clients.remove(client); + } + if (!client.getName().equals(messsage.getSenderName())) { client.send(messsage); @@ -78,15 +89,50 @@ private void broadcast() } } } + date = new Date(); lastUpdateTime = date.getTime(); } - Thread.sleep(2000); + Thread.sleep(500); } } catch (InterruptedException e) { throw new RuntimeException(e); } } + private void adminConsole() + { + Scanner scanner = new Scanner(System.in); + + while (true) { + String line = scanner.nextLine(); + if ("exit".equals(line)) { + System.out.println("bye!"); + break; + } + + if ("list".equals(line)) { + synchronized (clients){ + System.out.println("Client list:"); + if (clients.isEmpty()) { + System.out.println("empty"); + } + + for (ClientThread client: clients){ + if (!client.isAlive()) { + System.out.print("-"); + clients.remove(client); + } else { + System.out.print("+"); + } + + System.out.println(client.getName()); + + } + } + } + } + } + public static void main(String[] args) throws Exception { Server server = new Server(9000, 10); server.serve(); From 04a9380b7a09d688ce0107964cae949c23a142d1 Mon Sep 17 00:00:00 2001 From: Highoc Date: Wed, 6 Jun 2018 12:25:19 +0300 Subject: [PATCH 5/6] server without db --- .../java/ru/track/homework/AdminThread.java | 51 +++++++++++ .../ru/track/homework/BroadcosterThread.java | 68 ++++++++++++++ .../main/java/ru/track/homework/Client.java | 22 +++-- .../java/ru/track/homework/ClientThread.java | 54 +++++++---- .../track/homework/ConversationService.java | 36 ++++++++ .../main/java/ru/track/homework/Database.java | 25 +++++ .../main/java/ru/track/homework/Message.java | 17 ++-- .../main/java/ru/track/homework/Server.java | 91 +++---------------- 8 files changed, 247 insertions(+), 117 deletions(-) create mode 100644 L7_threads/src/main/java/ru/track/homework/AdminThread.java create mode 100644 L7_threads/src/main/java/ru/track/homework/BroadcosterThread.java create mode 100644 L7_threads/src/main/java/ru/track/homework/ConversationService.java create mode 100644 L7_threads/src/main/java/ru/track/homework/Database.java diff --git a/L7_threads/src/main/java/ru/track/homework/AdminThread.java b/L7_threads/src/main/java/ru/track/homework/AdminThread.java new file mode 100644 index 0000000..d5af1be --- /dev/null +++ b/L7_threads/src/main/java/ru/track/homework/AdminThread.java @@ -0,0 +1,51 @@ +package ru.track.homework; + +import org.jetbrains.annotations.NotNull; + +import java.util.List; +import java.util.Scanner; + +public class AdminThread extends Thread { + private List clients; + + public AdminThread(@NotNull List clients) { + this.clients = clients; + } + + public void run() { + Scanner scanner = new Scanner(System.in); + + while (!isInterrupted()) { + String line = scanner.next(); + + if (line.equals("list")) { + synchronized (clients){ + System.out.println("Client list:"); + + if (clients.isEmpty()) { + System.out.println("empty"); + continue; + } + + for (ClientThread client: clients){ + if (!client.isInterrupted()) { + System.out.println(String.format("%s[%s]", client.getName(), client.getUsername())); + } + } + } + } + else if (line.equals("drop")) { + long clientId = scanner.nextLong(); + + synchronized (clients){ + for (ClientThread client: clients){ + if (!client.isInterrupted() && (client.getClientID() == clientId)) { + client.interrupt(); + } + } + } + + } + } + } +} \ No newline at end of file diff --git a/L7_threads/src/main/java/ru/track/homework/BroadcosterThread.java b/L7_threads/src/main/java/ru/track/homework/BroadcosterThread.java new file mode 100644 index 0000000..869f75c --- /dev/null +++ b/L7_threads/src/main/java/ru/track/homework/BroadcosterThread.java @@ -0,0 +1,68 @@ +package ru.track.homework; + +import org.jetbrains.annotations.NotNull; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; + +public class BroadcosterThread extends Thread { + private static List messages; + private static List clients; + + private static long lastUpdateTS = System.currentTimeMillis(); + + public BroadcosterThread(@NotNull List clients, @NotNull List messages) { + this.messages = messages; + this.clients = clients; + } + + public void run() + { + try { + while (!isInterrupted()) { + List newMessages = getNewMessages(); + for (Message newMessage: newMessages) { + synchronized (clients) { + for (ClientThread client: clients) { + + if (client.isInterrupted()) { + clients.remove(client); + continue; + } + + if (!client.getUsername().equals(newMessage.getUsername())) { + try { + client.send(newMessage); + } catch (IOException e) { + client.interrupt(); + clients.remove(client); + } + } + } + } + } + + Thread.sleep(800); + } + } catch (InterruptedException e) { + //throw new RuntimeException(e); + } + } + + private List getNewMessages() { + long nowTS = System.currentTimeMillis(); + List result = new ArrayList<>(); + + synchronized (messages) { + for (Message messsage: messages) { + if (messsage.getTimeStand() > lastUpdateTS) { + result.add(messsage); + } + } + } + lastUpdateTS = nowTS; + + return result; + } +} diff --git a/L7_threads/src/main/java/ru/track/homework/Client.java b/L7_threads/src/main/java/ru/track/homework/Client.java index bc09bd4..205d308 100644 --- a/L7_threads/src/main/java/ru/track/homework/Client.java +++ b/L7_threads/src/main/java/ru/track/homework/Client.java @@ -8,14 +8,14 @@ import java.io.InputStream; import java.io.OutputStream; import java.net.Socket; -import java.util.Date; import java.util.Scanner; public class Client { - private final int port; private final String host; + private String username; + private Thread writer; private Thread reader; @@ -33,15 +33,13 @@ private void connect() final Socket finalSocket = socket; writer = new Thread(() -> { writeToSocket(finalSocket); }); - writer.setName("writer"); writer.start(); reader = new Thread(() -> { readFromSocket(finalSocket); }); - reader.setName("reader"); reader.start(); } catch (IOException e) { - throw new RuntimeException(e); + //throw new RuntimeException(e); } } @@ -54,12 +52,12 @@ private void readFromSocket(@NotNull Socket socket) { if (nRead == 0) continue; Message message = gson.fromJson(new String(buffer, 0, nRead), Message.class); - System.out.println(String.format("%s> %s", message.getSenderName(), message.getData())); + System.out.println(String.format("%s> %s", message.getUsername(), message.getText())); } socket.shutdownInput(); } catch (IOException e) { - throw new RuntimeException(e); + //throw new RuntimeException(e); } finally { IOUtils.closeQuietly(socket); } @@ -70,6 +68,11 @@ private void writeToSocket(@NotNull Socket socket) { final OutputStream out = socket.getOutputStream(); Scanner scanner = new Scanner(System.in); + System.out.print("Введите ваш username: "); + username = scanner.nextLine(); + + out.write(username.getBytes()); + while (true) { String line = scanner.nextLine(); if ("exit".equals(line)) { @@ -77,8 +80,7 @@ private void writeToSocket(@NotNull Socket socket) { break; } - Date date = new Date(); - Message message = new Message("", line, date.getTime()); + Message message = new Message(username, line); String result = gson.toJson(message); out.write(result.getBytes()); @@ -87,7 +89,7 @@ private void writeToSocket(@NotNull Socket socket) { socket.shutdownOutput(); } catch (IOException e) { - throw new RuntimeException(e); + //throw new RuntimeException(e); } finally { IOUtils.closeQuietly(socket); } diff --git a/L7_threads/src/main/java/ru/track/homework/ClientThread.java b/L7_threads/src/main/java/ru/track/homework/ClientThread.java index 6fc9130..363b5d7 100644 --- a/L7_threads/src/main/java/ru/track/homework/ClientThread.java +++ b/L7_threads/src/main/java/ru/track/homework/ClientThread.java @@ -13,13 +13,17 @@ import java.util.Date; import java.util.List; import java.util.Queue; +import java.util.concurrent.atomic.AtomicLong; public class ClientThread extends Thread { + private final long clientID; + private String username; - private final int clientID; - private static int counterID = 0; + private static AtomicCounter counter = new AtomicCounter(); private static List messages; + private static List clients; + private final Socket socket; private InputStream in; @@ -27,49 +31,59 @@ public class ClientThread extends Thread { private static Gson gson = new Gson(); - public ClientThread(@NotNull Socket socket, @NotNull List messages) { - this.clientID = counterID++; + public ClientThread(@NotNull Socket socket, @NotNull List clients, @NotNull List messages) { + this.clientID = counter.inc(); this.socket = socket; + + this.clients = clients; this.messages = messages; + setName(String.format("Client[%d]@%s:%s", clientID, socket.getInetAddress(), socket.getPort())); } - public void run() - { + public void run() { try { in = socket.getInputStream(); out = socket.getOutputStream(); final byte[] buffer = new byte[4096]; - for (int nRead = 0; nRead != -1; nRead = in.read(buffer)) { + int nUsername = in.read(buffer); + username = new String(buffer, 0, nUsername); + + for (int nRead = 0; !isInterrupted() || (nRead != -1); nRead = in.read(buffer)) { + while (in.available() == 0) { sleep(200); } if (nRead == 0) continue; Message message = gson.fromJson(new String(buffer, 0, nRead), Message.class); - message.setSenderName(getName()); messages.add(message); - System.out.println(String.format("%s> %s", message.getSenderName(), message.getData())); + System.out.println(String.format("%s[%s]> %s", getName(), message.getUsername(), message.getText())); } socket.shutdownInput(); socket.shutdownOutput(); - } catch (IOException e) { - throw new RuntimeException(e); + } catch (IOException | InterruptedException e) { + //throw new RuntimeException(e); } finally { + clients.remove(this); System.out.println(String.format("main> client disconnected: %s", getName())); IOUtils.closeQuietly(socket); } } - public void send(Message message) - { - try { - String result = gson.toJson(message); - out.write(result.getBytes()); - out.flush(); - } catch (IOException e) { - throw new RuntimeException(e); - } + public void send(Message message) throws IOException { + out.write(gson.toJson(message).getBytes()); + out.flush(); + } + + public String getUsername() { return username; } + public long getClientID() { return clientID; } +} + +class AtomicCounter { + private AtomicLong val = new AtomicLong(0); + public long inc() { + return val.getAndIncrement(); } } diff --git a/L7_threads/src/main/java/ru/track/homework/ConversationService.java b/L7_threads/src/main/java/ru/track/homework/ConversationService.java new file mode 100644 index 0000000..4026f6b --- /dev/null +++ b/L7_threads/src/main/java/ru/track/homework/ConversationService.java @@ -0,0 +1,36 @@ +package ru.track.homework; +import java.util.List; + +public interface ConversationService { + /** + * В зависимости от Message.senderName нужно сохранять в разные базы + * + * @return вернуть ID, который был присвоен сообщению в базе (поле ID) + * + * + * 4 балла + */ + long store(Message msg); + + /** + * Получить историю сообщений за период времени. Важно учесть лимит, чтобы не свалить базы слишком большой выборкой. + * Ограничить LIMIT нужно именно при запросе в базу + * @param from - timestamp с какого времени + * @param to - timestamp до какого времени + * @param limit - максимальное кол-во ссобщений + * + * @return Список, отсротированный по timestamp + * + * 4 балла + */ + List getHistory(long from, long to, long limit); + + + /** + * Вернуть все сообщения от определенного пользователя + * + * 4 балла + */ + List getByUser(String username, long limit); + +} diff --git a/L7_threads/src/main/java/ru/track/homework/Database.java b/L7_threads/src/main/java/ru/track/homework/Database.java new file mode 100644 index 0000000..23726d0 --- /dev/null +++ b/L7_threads/src/main/java/ru/track/homework/Database.java @@ -0,0 +1,25 @@ +package ru.track.homework; + +import java.util.List; + +public class Database implements ConversationService { + + public Database() { + + } + + @Override + public synchronized long store(Message msg) { + return 0; + } + + @Override + public synchronized List getHistory(long from, long to, long limit) { + return null; + } + + @Override + public synchronized List getByUser(String username, long limit) { + return null; + } +} diff --git a/L7_threads/src/main/java/ru/track/homework/Message.java b/L7_threads/src/main/java/ru/track/homework/Message.java index 6c81285..4e2ea05 100644 --- a/L7_threads/src/main/java/ru/track/homework/Message.java +++ b/L7_threads/src/main/java/ru/track/homework/Message.java @@ -3,19 +3,18 @@ import org.jetbrains.annotations.NotNull; class Message { - private String senderName; - private String data; + private String username; + private String text; private long ts; - public Message(@NotNull String senderName, @NotNull String data, long ts) + public Message(@NotNull String username, @NotNull String text) { - this.senderName = senderName; - this.ts = ts; - this.data = data; + this.username = username; + this.ts = System.currentTimeMillis(); + this.text = text; } - public String getData() { return data; } - public String getSenderName() { return senderName; } - public void setSenderName(@NotNull String senderName) { this.senderName = senderName; } + public String getText() { return text; } + public String getUsername() { return username; } public long getTimeStand() { return ts; } } \ No newline at end of file diff --git a/L7_threads/src/main/java/ru/track/homework/Server.java b/L7_threads/src/main/java/ru/track/homework/Server.java index ac2bd92..7579a7d 100644 --- a/L7_threads/src/main/java/ru/track/homework/Server.java +++ b/L7_threads/src/main/java/ru/track/homework/Server.java @@ -14,22 +14,24 @@ import java.net.Socket; import java.util.*; -//Не пробрасывать id на клиент -//Корректная работа со списком клиентов -//Админская консоль - public class Server { private final int port; private final int backlog; private List messages = Collections.synchronizedList(new ArrayList<>()); private List clients = Collections.synchronizedList(new ArrayList<>()); - private Thread broadcasterThread = new Thread(() -> { broadcast(); }); - private Thread adminConsoleThread = new Thread(() -> { adminConsole(); }); + + private BroadcosterThread broadcaster; + private AdminThread admin; + + private ConversationService storage = new Database(); public Server(int port, int backlog) { this.port = port; this.backlog = backlog; + + admin = new AdminThread(clients); + broadcaster = new BroadcosterThread(clients, messages); } public void serve() throws IOException { @@ -38,8 +40,8 @@ public void serve() throws IOException { final ServerSocket serverSocketFinal = new ServerSocket(port, backlog, InetAddress.getByName("localhost")); serverSocket = serverSocketFinal; - broadcasterThread.start(); - adminConsoleThread.start(); + broadcaster.start(); + admin.start(); while (true){ System.out.println("main> wait to accept new client..."); @@ -55,81 +57,14 @@ private void handle(@NotNull ServerSocket serverSocket) { try { Socket socket = serverSocket.accept(); - ClientThread client = new ClientThread(socket, messages); + ClientThread client = new ClientThread(socket, clients, messages); + clients.add(client); client.start(); System.out.println(String.format("main> new client connected: %s", client.getName())); } catch (IOException e) { - throw new RuntimeException(e); - } - } - - private void broadcast() - { - BufferedInputStream - try { - Date date = new Date(); - long lastUpdateTime = date.getTime(); - - while (true) { - synchronized (messages) { - for (Message messsage: messages) { - if (messsage.getTimeStand() >= lastUpdateTime) - { - for (ClientThread client: clients) { - if (!client.isAlive()) { - clients.remove(client); - } - - if (!client.getName().equals(messsage.getSenderName())) - { - client.send(messsage); - } - } - } - } - date = new Date(); - lastUpdateTime = date.getTime(); - } - Thread.sleep(500); - } - } catch (InterruptedException e) { - throw new RuntimeException(e); - } - } - - private void adminConsole() - { - Scanner scanner = new Scanner(System.in); - - while (true) { - String line = scanner.nextLine(); - if ("exit".equals(line)) { - System.out.println("bye!"); - break; - } - - if ("list".equals(line)) { - synchronized (clients){ - System.out.println("Client list:"); - if (clients.isEmpty()) { - System.out.println("empty"); - } - - for (ClientThread client: clients){ - if (!client.isAlive()) { - System.out.print("-"); - clients.remove(client); - } else { - System.out.print("+"); - } - - System.out.println(client.getName()); - - } - } - } + //throw new RuntimeException(e); } } From d61f0bd35d6758ba8649f47ea9126b2e97ee7bb0 Mon Sep 17 00:00:00 2001 From: Highoc Date: Wed, 6 Jun 2018 15:35:24 +0300 Subject: [PATCH 6/6] with DB --- L7_threads/pom.xml | 8 ++ .../java/ru/track/homework/AdminThread.java | 3 +- .../ru/track/homework/BroadcosterThread.java | 39 ++----- .../java/ru/track/homework/ClientThread.java | 15 ++- .../main/java/ru/track/homework/Database.java | 106 +++++++++++++++++- .../main/java/ru/track/homework/Message.java | 9 +- .../main/java/ru/track/homework/Server.java | 8 +- 7 files changed, 134 insertions(+), 54 deletions(-) diff --git a/L7_threads/pom.xml b/L7_threads/pom.xml index 2bb00c4..699959b 100644 --- a/L7_threads/pom.xml +++ b/L7_threads/pom.xml @@ -101,5 +101,13 @@ 2.8.2 + + + mysql + mysql-connector-java + 5.1.6 + + + \ No newline at end of file diff --git a/L7_threads/src/main/java/ru/track/homework/AdminThread.java b/L7_threads/src/main/java/ru/track/homework/AdminThread.java index d5af1be..eb9785d 100644 --- a/L7_threads/src/main/java/ru/track/homework/AdminThread.java +++ b/L7_threads/src/main/java/ru/track/homework/AdminThread.java @@ -8,7 +8,7 @@ public class AdminThread extends Thread { private List clients; - public AdminThread(@NotNull List clients) { + AdminThread(@NotNull List clients) { this.clients = clients; } @@ -41,6 +41,7 @@ else if (line.equals("drop")) { for (ClientThread client: clients){ if (!client.isInterrupted() && (client.getClientID() == clientId)) { client.interrupt(); + break; } } } diff --git a/L7_threads/src/main/java/ru/track/homework/BroadcosterThread.java b/L7_threads/src/main/java/ru/track/homework/BroadcosterThread.java index 869f75c..7f2628a 100644 --- a/L7_threads/src/main/java/ru/track/homework/BroadcosterThread.java +++ b/L7_threads/src/main/java/ru/track/homework/BroadcosterThread.java @@ -3,40 +3,35 @@ import org.jetbrains.annotations.NotNull; import java.io.IOException; -import java.util.ArrayList; import java.util.List; public class BroadcosterThread extends Thread { - private static List messages; + private static ConversationService messages; private static List clients; private static long lastUpdateTS = System.currentTimeMillis(); - public BroadcosterThread(@NotNull List clients, @NotNull List messages) { - this.messages = messages; - this.clients = clients; + BroadcosterThread(@NotNull List clients, @NotNull ConversationService messages) { + BroadcosterThread.messages = messages; + BroadcosterThread.clients = clients; } public void run() { try { while (!isInterrupted()) { - List newMessages = getNewMessages(); + long currentTS = System.currentTimeMillis(); + List newMessages = messages.getHistory(lastUpdateTS, currentTS, 20); + lastUpdateTS = currentTS; + for (Message newMessage: newMessages) { synchronized (clients) { for (ClientThread client: clients) { - - if (client.isInterrupted()) { - clients.remove(client); - continue; - } - - if (!client.getUsername().equals(newMessage.getUsername())) { + if (!client.isInterrupted() && !client.getUsername().equals(newMessage.getUsername())) { try { client.send(newMessage); } catch (IOException e) { client.interrupt(); - clients.remove(client); } } } @@ -49,20 +44,4 @@ public void run() //throw new RuntimeException(e); } } - - private List getNewMessages() { - long nowTS = System.currentTimeMillis(); - List result = new ArrayList<>(); - - synchronized (messages) { - for (Message messsage: messages) { - if (messsage.getTimeStand() > lastUpdateTS) { - result.add(messsage); - } - } - } - lastUpdateTS = nowTS; - - return result; - } } diff --git a/L7_threads/src/main/java/ru/track/homework/ClientThread.java b/L7_threads/src/main/java/ru/track/homework/ClientThread.java index 363b5d7..7d82f62 100644 --- a/L7_threads/src/main/java/ru/track/homework/ClientThread.java +++ b/L7_threads/src/main/java/ru/track/homework/ClientThread.java @@ -21,7 +21,7 @@ public class ClientThread extends Thread { private static AtomicCounter counter = new AtomicCounter(); - private static List messages; + private static ConversationService messages; private static List clients; private final Socket socket; @@ -31,12 +31,12 @@ public class ClientThread extends Thread { private static Gson gson = new Gson(); - public ClientThread(@NotNull Socket socket, @NotNull List clients, @NotNull List messages) { + ClientThread(@NotNull Socket socket, @NotNull List clients, @NotNull ConversationService messages) { this.clientID = counter.inc(); this.socket = socket; - this.clients = clients; - this.messages = messages; + ClientThread.clients = clients; + ClientThread.messages = messages; setName(String.format("Client[%d]@%s:%s", clientID, socket.getInetAddress(), socket.getPort())); } @@ -50,20 +50,19 @@ public void run() { int nUsername = in.read(buffer); username = new String(buffer, 0, nUsername); - for (int nRead = 0; !isInterrupted() || (nRead != -1); nRead = in.read(buffer)) { - while (in.available() == 0) { sleep(200); } + //while (in.available() == 0) { sleep(200); } if (nRead == 0) continue; Message message = gson.fromJson(new String(buffer, 0, nRead), Message.class); - messages.add(message); System.out.println(String.format("%s[%s]> %s", getName(), message.getUsername(), message.getText())); + messages.store(message); } socket.shutdownInput(); socket.shutdownOutput(); - } catch (IOException | InterruptedException e) { + } catch (IOException e) { //| InterruptedException e //throw new RuntimeException(e); } finally { clients.remove(this); diff --git a/L7_threads/src/main/java/ru/track/homework/Database.java b/L7_threads/src/main/java/ru/track/homework/Database.java index 23726d0..b889c2c 100644 --- a/L7_threads/src/main/java/ru/track/homework/Database.java +++ b/L7_threads/src/main/java/ru/track/homework/Database.java @@ -1,25 +1,121 @@ package ru.track.homework; +import java.sql.*; +import java.util.ArrayList; import java.util.List; public class Database implements ConversationService { - public Database() { + private Connection[] connections = new Connection[3]; + Database() { + try { + DriverManager.registerDriver((Driver) Class.forName("com.mysql.jdbc.Driver").newInstance()); + connections[0] = connect(1); + connections[1] = connect(2); + connections[2] = connect(3); + } catch (SQLException |IllegalAccessException | ClassNotFoundException | InstantiationException e) { + throw new RuntimeException(e); + } } @Override public synchronized long store(Message msg) { - return 0; + Connection currentConnection = getConnection(msg.getUsername()); + + try { + String insert = "INSERT INTO messages (user_name, text, ts) VALUES (?, ?, ?)"; + PreparedStatement stmt = currentConnection.prepareStatement(insert, Statement.RETURN_GENERATED_KEYS); + stmt.setString(1, msg.getUsername()); + stmt.setString(2, msg.getText()); + stmt.setTimestamp(3, new java.sql.Timestamp(msg.getTimeStand())); + + int affectedRows = stmt.executeUpdate(); + + if (affectedRows == 0) { + throw new SQLException("Storing message failed, no rows affected."); + } + + try (ResultSet generatedKeys = stmt.getGeneratedKeys()) { + if (generatedKeys.next()) { + return generatedKeys.getLong(1); + } + else { + throw new SQLException("Storing message failed, no ID obtained."); + } + } + } catch (SQLException e) { + throw new RuntimeException(e); + } } @Override public synchronized List getHistory(long from, long to, long limit) { - return null; + List result = new ArrayList<>(); + + try { + for (Connection currentConnection : connections) { + String select = "SELECT * FROM messages WHERE ts >= ? AND ts < ? ORDER BY ts LIMIT ?"; + PreparedStatement stmt = currentConnection.prepareStatement(select); + stmt.setTimestamp(1, new java.sql.Timestamp(from)); + stmt.setTimestamp(2, new java.sql.Timestamp(to)); + stmt.setLong(3, limit); + + ResultSet resultSet = stmt.executeQuery(); + while (resultSet.next()) { + result.add(new Message(resultSet.getString(2), resultSet.getString(3), resultSet.getLong(4))); + } + } + + result.sort((a, b) -> (int) (a.getTimeStand() - b.getTimeStand())); + + } catch (SQLException e) { + throw new RuntimeException(e); + } + + return result; } @Override public synchronized List getByUser(String username, long limit) { - return null; + Connection currentConnection = getConnection(username); + List result = new ArrayList<>(); + + try { + String select = "SELECT * FROM messages WHERE user_name = '?' LIMIT ? ORDER BY ts"; + PreparedStatement stmt = currentConnection.prepareStatement(select); + stmt.setString(1, username); + stmt.setLong(2, limit); + + ResultSet resultSet = stmt.executeQuery(); + while(resultSet.next()) { + result.add(new Message(resultSet.getString(2), resultSet.getString(3), resultSet.getLong(4))); + } + + } catch (SQLException e) { + throw new RuntimeException(e); + } + + return result; + } + + private static Connection connect(int shardID) throws SQLException { + String url = String.format("jdbc:mysql://tdb-%d.trail5.net:3306/track17?user=track_student&password=7EsH.H6x", shardID); + return DriverManager.getConnection(url); + } + + private Connection getConnection(String username) { + char firstChar = Character.toUpperCase(username.charAt(0)); + + Connection currentConnection; + if (('A' <= firstChar) && (firstChar <= 'J')) + currentConnection = connections[0]; + else if (('K' <= firstChar) && (firstChar <= 'T')) + currentConnection = connections[1]; + else if (('U' <= firstChar) && (firstChar <= 'Z')) + currentConnection = connections[2]; + else currentConnection = null; + + return currentConnection; } -} +} \ No newline at end of file diff --git a/L7_threads/src/main/java/ru/track/homework/Message.java b/L7_threads/src/main/java/ru/track/homework/Message.java index 4e2ea05..116b87d 100644 --- a/L7_threads/src/main/java/ru/track/homework/Message.java +++ b/L7_threads/src/main/java/ru/track/homework/Message.java @@ -7,10 +7,13 @@ class Message { private String text; private long ts; - public Message(@NotNull String username, @NotNull String text) - { + Message(@NotNull String username, @NotNull String text) { + this(username, text, System.currentTimeMillis()); + } + + Message(@NotNull String username, @NotNull String text, long ts) { this.username = username; - this.ts = System.currentTimeMillis(); + this.ts = ts; this.text = text; } diff --git a/L7_threads/src/main/java/ru/track/homework/Server.java b/L7_threads/src/main/java/ru/track/homework/Server.java index 7579a7d..576feef 100644 --- a/L7_threads/src/main/java/ru/track/homework/Server.java +++ b/L7_threads/src/main/java/ru/track/homework/Server.java @@ -2,13 +2,8 @@ import org.apache.commons.io.IOUtils; import org.jetbrains.annotations.NotNull; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import java.io.BufferedInputStream; import java.io.IOException; -import java.io.InputStream; -import java.io.OutputStream; import java.net.InetAddress; import java.net.ServerSocket; import java.net.Socket; @@ -18,13 +13,12 @@ public class Server { private final int port; private final int backlog; - private List messages = Collections.synchronizedList(new ArrayList<>()); + private ConversationService messages = new Database(); private List clients = Collections.synchronizedList(new ArrayList<>()); private BroadcosterThread broadcaster; private AdminThread admin; - private ConversationService storage = new Database(); public Server(int port, int backlog) { this.port = port;