From 0f170b86915753c42f603bf92de86f577f6ece2f Mon Sep 17 00:00:00 2001 From: Ivan Zinchenko Date: Wed, 5 Jun 2024 15:47:20 +0300 Subject: [PATCH] =?UTF-8?q?=D0=94=D0=BE=D0=B1=D0=B0=D0=B2=D0=BB=D0=B5?= =?UTF-8?q?=D0=BD=D0=B0=20=D0=BC=D0=BD=D0=BE=D0=B3=D0=BE=D0=BF=D0=BE=D1=82?= =?UTF-8?q?=D0=BE=D1=87=D0=BD=D0=BE=D1=81=D1=82=D1=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../me/zinch/Lab7/Server/Server/Server.java | 142 +++++++-------- .../Server/wrapper/ProductCollection.java | 166 +++++++++++++----- 2 files changed, 186 insertions(+), 122 deletions(-) diff --git a/Lab.Server/src/main/java/me/zinch/Lab7/Server/Server/Server.java b/Lab.Server/src/main/java/me/zinch/Lab7/Server/Server/Server.java index aaa31a1..2cfaf69 100644 --- a/Lab.Server/src/main/java/me/zinch/Lab7/Server/Server/Server.java +++ b/Lab.Server/src/main/java/me/zinch/Lab7/Server/Server/Server.java @@ -18,34 +18,28 @@ import org.slf4j.LoggerFactory; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; import java.io.IOException; +import java.io.InputStream; import java.io.ObjectInputStream; import java.io.ObjectOutputStream; import java.io.Serializable; -import java.net.InetSocketAddress; -import java.net.SocketAddress; -import java.nio.ByteBuffer; -import java.nio.channels.SelectionKey; -import java.nio.channels.Selector; -import java.nio.channels.ServerSocketChannel; -import java.nio.channels.SocketChannel; +import java.net.ServerSocket; +import java.net.Socket; import java.util.ArrayList; -import java.util.HashMap; -import java.util.Map; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; public class Server { private static final Logger log = LoggerFactory.getLogger(Server.class); - private final Selector selector; - private final ServerSocketChannel serverSocketChannel; - private final Map usersMessages = new HashMap<>(); + private final ServerSocket serverSocket; private final IStorage storage; - private boolean isRunning = false; + private boolean isRunning; + private final ExecutorService requestReadPool = Executors.newCachedThreadPool(); + private final ExecutorService requestProcessPool = Executors.newCachedThreadPool(); + private final ExecutorService responseSendPool = Executors.newFixedThreadPool(12); public Server(int port, IStorage collection) throws IOException { - selector = Selector.open(); - serverSocketChannel = ServerSocketChannel.open(); - serverSocketChannel.bind(new InetSocketAddress(port)); - serverSocketChannel.configureBlocking(false); - serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT); + serverSocket = new ServerSocket(port); isRunning = true; if (collection == null) collection = new ProductCollection(new ArrayList<>()); @@ -54,80 +48,70 @@ public class Server { log.info("The server has been assigned to port {}", port); } - public static ByteBuffer serialize(Serializable obj) throws IOException { + public static byte[] serialize(Serializable obj) throws IOException { try (var bOut = new ByteArrayOutputStream(); var oOut = new ObjectOutputStream(bOut)) { oOut.writeObject(obj); oOut.flush(); - return ByteBuffer.wrap(bOut.toByteArray()); + return bOut.toByteArray(); } } - public void run() throws IOException { + public void run() { log.info("The server is running"); while (isRunning) { - selector.select(this::handleKey); + try { + var clientSocket = serverSocket.accept(); + log.info("Received connection from {}", clientSocket.getRemoteSocketAddress()); + requestReadPool.execute(() -> { + try { + handleClient(clientSocket); + } catch (IOException | ClassNotFoundException e) { + log.error(e.getMessage(), e); + } + }); + } catch (IOException e) { + log.error(e.getMessage(), e); + } } } - private void handleKey(SelectionKey key) { + private void handleClient(Socket clientSocket) throws IOException, ClassNotFoundException { + var inputStream = clientSocket.getInputStream(); + + var buffer = new byte[8192]; + inputStream.read(buffer); + + var oIn = new ObjectInputStream(new ByteArrayInputStream(buffer)); + var msg = (Message) oIn.readObject(); + log.info("Received message {} from {}", msg, clientSocket.getRemoteSocketAddress()); + + responseSendPool.execute(() -> handleWrite(clientSocket, msg, inputStream)); + } + + private void handleWrite(Socket clientSocket, Message msg, InputStream inputStream) { try { - if (key.isAcceptable()) handleAccept(key); - if (key.isReadable()) handleRead(key); - if (key.isWritable()) handleWrite(key); - } catch (IOException | ClassNotFoundException e) { + var response = requestProcessPool.submit(() -> createResponse(msg)); + var responseBuffer = response.get(); + if (responseBuffer != null) { + try { + var outputStream = clientSocket.getOutputStream(); + outputStream.write(responseBuffer); + outputStream.flush(); + log.info("Sent message to {}", clientSocket.getRemoteSocketAddress()); + outputStream.close(); + } catch (IOException e) { + log.error(e.getMessage(), e); + } + } + inputStream.close(); + } catch (IOException | InterruptedException | ExecutionException e) { log.error(e.getMessage(), e); - key.cancel(); } } - private void handleAccept(SelectionKey key) throws IOException { - var server = (ServerSocketChannel) key.channel(); - var client = server.accept(); - if (client != null) { - client.configureBlocking(false); - client.register(selector, SelectionKey.OP_READ); - log.info("Received connection from {}", client.getRemoteAddress()); - } - } - - private void handleRead(SelectionKey key) throws IOException, ClassNotFoundException { - var client = (SocketChannel) key.channel(); - var buffer = ByteBuffer.allocate(8192); - int bytesRead = client.read(buffer); - if (bytesRead == -1) { - client.close(); - } else { - buffer.flip(); - var oIn = new ObjectInputStream(new ByteArrayInputStream(buffer.array())); - var msg = (Message) oIn.readObject(); - log.info("Received message {} from {}", msg, client.getRemoteAddress()); - usersMessages.put(client.getRemoteAddress(), msg); - client.register(selector, SelectionKey.OP_WRITE); - oIn.close(); - } - } - - private void handleWrite(SelectionKey key) throws IOException { - var client = (SocketChannel) key.channel(); - var message = usersMessages.get(client.getRemoteAddress()); - if (message == null) { - log.error("Message for {} not found", client.getRemoteAddress()); - client.close(); - return; - } - - ByteBuffer responseBuffer = createResponse(message); - if (responseBuffer != null) { - client.write(responseBuffer); - log.info("Sent message to {}", client.getRemoteAddress()); - } - - client.close(); - } - - private ByteBuffer createResponse(Message message) throws IOException { + private byte[] createResponse(Message message) throws IOException { var user = (User) message.getUser(); if (user != null) user.setPassword(SHA256.hash(user.getPassword())); return switch (message.getType()) { @@ -141,7 +125,7 @@ public class Server { }; } - private ByteBuffer handleGetCommand(String inputCommand, User user) throws IOException { + private byte[] handleGetCommand(String inputCommand, User user) throws IOException { var command = CommandManager.getCommandByInput(inputCommand); if (command.isEmpty()) { log.error("Command {} not found", inputCommand); @@ -155,7 +139,7 @@ public class Server { } } - private ByteBuffer handlePostCommand(MessageBody request, User user) throws IOException { + private byte[] handlePostCommand(MessageBody request, User user) throws IOException { var inputCommand = request.getCommand(); var body = request.getBody(); @@ -175,7 +159,9 @@ public class Server { public void stop() throws IOException { isRunning = false; - serverSocketChannel.close(); - selector.close(); + serverSocket.close(); + requestReadPool.shutdown(); + requestProcessPool.shutdown(); + responseSendPool.shutdown(); } } diff --git a/Lab.Server/src/main/java/me/zinch/Lab7/Server/wrapper/ProductCollection.java b/Lab.Server/src/main/java/me/zinch/Lab7/Server/wrapper/ProductCollection.java index 4d5715f..9cf2f0a 100644 --- a/Lab.Server/src/main/java/me/zinch/Lab7/Server/wrapper/ProductCollection.java +++ b/Lab.Server/src/main/java/me/zinch/Lab7/Server/wrapper/ProductCollection.java @@ -20,14 +20,13 @@ import java.util.Objects; import java.util.Optional; import java.util.Set; import java.util.TreeSet; +import java.util.concurrent.locks.ReentrantLock; +import java.util.concurrent.locks.Lock; -/** - * Represents a collection of products. - */ public class ProductCollection implements IStorage { private static final Logger log = LoggerFactory.getLogger(ProductCollection.class); private TreeSet productList; - private Long idIncrementor; + private final Lock lock = new ReentrantLock(); private List fetchProductFromDb(Connection connection) { try { @@ -45,7 +44,6 @@ public class ProductCollection implements IStorage { return Math.toIntExact((long) (o1Length - o2Length)); }); productList.addAll(list); - idIncrementor = productList.isEmpty() ? 1L : productList.stream().map(Product::getId).max(Long::compare).orElse(1L); } public ProductCollection(List list) { @@ -61,84 +59,154 @@ public class ProductCollection implements IStorage { } public Product getProductById(Long id) { - Optional product = productList.stream().filter(i -> Objects.equals(i.getId(), id)).findFirst(); - return product.orElse(null); + lock.lock(); + try { + Optional product = productList.stream().filter(i -> Objects.equals(i.getId(), id)).findFirst(); + return product.orElse(null); + } finally { + lock.unlock(); + } } public Product addProduct(ProductDTO productDTO, User user) throws SQLException { - var product = DbWrapper.addProduct(productDTO, DbController.getConnection(), user); - productList.add(product); - return product; + lock.lock(); + try { + var product = DbWrapper.addProduct(productDTO, DbController.getConnection(), user); + productList.add(product); + return product; + } finally { + lock.unlock(); + } } public Product updateProduct(Long id, ProductDTO productDTO, User user) throws SQLException { - var product = DbWrapper.updateProduct(id, productDTO, DbController.getConnection(), user); - fetchProductFromDb(DbController.getConnection()); - return productDTO.buildProduct(product.getId(), product.getCreationDate()); + lock.lock(); + try { + var product = DbWrapper.updateProduct(id, productDTO, DbController.getConnection(), user); + fetchProductFromDb(DbController.getConnection()); + return productDTO.buildProduct(product.getId(), product.getCreationDate()); + } finally { + lock.unlock(); + } } public Product removeProduct(Long id, User user) throws SQLException { - var product = getProductById(id); - if (DbWrapper.removeProductById(id, DbController.getConnection(), user)) { - productList.remove(product); - return product; + lock.lock(); + try { + var product = getProductById(id); + if (DbWrapper.removeProductById(id, DbController.getConnection(), user)) { + productList.remove(product); + return product; + } + return null; + } finally { + lock.unlock(); } - return null; } public String getInfo() { - var optionalInitDate = productList.stream().map(Product::getCreationDate).sorted().findFirst(); - ZonedDateTime initDate = optionalInitDate.orElse(ZonedDateTime.ofInstant(Instant.EPOCH, ZoneId.systemDefault())); - return String.format("Структура: TreeSet%nДата инициализации: %s%nКоличество элементов: %s", initDate, productList.size()); + lock.lock(); + try { + var optionalInitDate = productList.stream().map(Product::getCreationDate).sorted().findFirst(); + ZonedDateTime initDate = optionalInitDate.orElse(ZonedDateTime.ofInstant(Instant.EPOCH, ZoneId.systemDefault())); + return String.format("Структура: TreeSet%nДата инициализации: %s%nКоличество элементов: %s", initDate, productList.size()); + } finally { + lock.unlock(); + } } public void clear(User user) throws SQLException { - DbWrapper.clearProducts(user, DbController.getConnection()); - fetchProductFromDb(DbController.getConnection()); + lock.lock(); + try { + DbWrapper.clearProducts(user, DbController.getConnection()); + fetchProductFromDb(DbController.getConnection()); + } finally { + lock.unlock(); + } } public Long getMaxPrice() { - return productList.stream().max(Comparator.comparingLong(Product::getPrice)).map(Product::getPrice).orElse(null); + lock.lock(); + try { + return productList.stream().max(Comparator.comparingLong(Product::getPrice)).map(Product::getPrice).orElse(null); + } finally { + lock.unlock(); + } } public Long getMinPrice() { - return productList.stream().min(Comparator.comparingLong(Product::getPrice)).map(Product::getPrice).orElse(null); + lock.lock(); + try { + return productList.stream().min(Comparator.comparingLong(Product::getPrice)).map(Product::getPrice).orElse(null); + } finally { + lock.unlock(); + } } public boolean isProductIdExists(Long id) { - return getProductById(id) != null; + lock.lock(); + try { + return getProductById(id) != null; + } finally { + lock.unlock(); + } } public Integer removeLover(Long id, User user) throws SQLException { - var size = productList.size(); - DbWrapper.removeLower(id, DbController.getConnection(), user); - fetchProductFromDb(DbController.getConnection()); - return size - productList.size(); + lock.lock(); + try { + var size = productList.size(); + DbWrapper.removeLower(id, DbController.getConnection(), user); + fetchProductFromDb(DbController.getConnection()); + return size - productList.size(); + } finally { + lock.unlock(); + } } public String filterContainsName(String name) { - return String.join("\n", productList.stream() - .filter(product -> product.getName().toLowerCase().contains(name.toLowerCase())) - .map(Product::toString) - .toList()); + lock.lock(); + try { + return String.join("\n", productList.stream() + .filter(product -> product.getName().toLowerCase().contains(name.toLowerCase())) + .map(Product::toString) + .toList()); + } finally { + lock.unlock(); + } } public String getUniqueManufactureCost() { - return String.join(", ", Set.copyOf(productList.stream() - .map(Product::getManufactureCost) - .toList()) - .stream() - .map(Object::toString) - .toList()); + lock.lock(); + try { + return String.join(", ", Set.copyOf(productList.stream() + .map(Product::getManufactureCost) + .toList()) + .stream() + .map(Object::toString) + .toList()); + } finally { + lock.unlock(); + } } public List toList() { - return productList.stream().toList(); + lock.lock(); + try { + return productList.stream().toList(); + } finally { + lock.unlock(); + } } @Override public String toString() { - return String.join("\n", productList.stream().map(Product::toString).toList()); + lock.lock(); + try { + return String.join("\n", productList.stream().map(Product::toString).toList()); + } finally { + lock.unlock(); + } } @Override @@ -146,11 +214,21 @@ public class ProductCollection implements IStorage { if (this == o) return true; if (o == null || getClass() != o.getClass()) return false; ProductCollection that = (ProductCollection) o; - return Objects.equals(productList, that.productList) && Objects.equals(idIncrementor, that.idIncrementor); + lock.lock(); + try { + return Objects.equals(productList, that.productList); + } finally { + lock.unlock(); + } } @Override public int hashCode() { - return Objects.hash(productList, idIncrementor); + lock.lock(); + try { + return Objects.hashCode(productList); + } finally { + lock.unlock(); + } } }