diff --git a/.gitignore b/.gitignore index b6065fb..b8174c2 100644 --- a/.gitignore +++ b/.gitignore @@ -24,4 +24,5 @@ hs_err_pid* *.iml .idea -target \ No newline at end of file +target +logs diff --git a/hw03-gc/Conclusions.md b/hw03-gc/Conclusions.md new file mode 100644 index 0000000..6bfba4c --- /dev/null +++ b/hw03-gc/Conclusions.md @@ -0,0 +1,52 @@ +# +**Статистика работы сборщиков мусора (java-11-openjdk)** + +Имя | Событие | Длительность,мс +--- | :---: | :---: +|**Serial**| +Copy | 4 | 474 | +MarkSweepCompact | 1 | 675 +|**Parallel**| +PS Scavenge | 3 | 195 +PS MarkSweep | 5 | 4496 +|**CMS**| +ParNew | 4 | 547 +ConcurrentMarkSweep | 40 | 155899 +|**G1**| +G1 Young Generation | 16 | 223 +G1 Old Generation | 1 | 296 + +**Статистика работы сборщиков мусора (java-14-oracle)** + +Имя | Событие | Длительность,мс +--- | :---: | :---: +|**Serial**| +Copy | 4 | 429 | +MarkSweepCompact | 1 | 461 +|**Parallel**| +PS Scavenge | 3 | 203 +PS MarkSweep | 3 | 2100 +|**G1**| +G1 Young Generation | 16 | 264 +G1 Old Generation | 1 | 314 + +**Время работы приложения при использовании разных сборщиков мусора и jdk** + +Сборщик мусора|Время работы приложения (java-11-openjdk), с |Время работы приложения (java-14-oracle), с +---| :---:| :---: +Serial | 281 | 370 +Parallel | 314 | 337 +CMS | 342 | - +G1 | 397 | 383 + + **Выводы:** + + - Для написанного приложения лучше всего показал себя G1 сборщик мусора. С G1 приложение работало + дольше, общая длительность пауз была меньше, чем с Serial, Parallel, CMS сборщиками; + - Минорные сборки G1 начинались раньше (при размере heap 20Mb, тогда как для других GC первые + сборки стартовали при размере heap больше 100Mb), выполнялись чаще (G1 имеет максимальное число + минорных сборок из рассмотренных сборщиков мусора) и были короткими (время на одну минорную сборку + минимально у G1, общая длительность минорных сборок меньше, чем у Serial и CMS сборщиков); + - Эффективность Parallel и CMS сборщиков мусора оказалась наименьшей: полных сборок мусора было больше, + приложение останавливалось чаще, время, затраченное на сборку мусора, значительно превышало аналогичное + значение для Serial и G1 сборщиков. diff --git a/hw03-gc/README.md b/hw03-gc/README.md new file mode 100644 index 0000000..5f638fd --- /dev/null +++ b/hw03-gc/README.md @@ -0,0 +1,19 @@ +# Сравнение разных сборщиков мусора + Написать приложение, которое следит за сборками мусора и пишет в лог количество сборок каждого типа + (young, old) и время которое ушло на сборки в минуту. + + Добиться OutOfMemory в этом приложении через медленное подтекание по памяти + (например добавлять элементы в List и удалять только половину). + + Настроить приложение (можно добавлять Thread.sleep(...)) так чтобы оно падало + с OOM примерно через 5 минут после начала работы. + + Собрать статистику (количество сборок, время на сборки) по разным GC. + + !!! Сделать выводы !!! + ЭТО САМАЯ ВАЖНАЯ ЧАСТЬ РАБОТЫ: + Какой gc лучше и почему? + + Выводы оформить в файле Сonclusions.md в корне папки проекта. + Результаты измерений сведите в таблицу. + \ No newline at end of file diff --git a/hw03-gc/pom.xml b/hw03-gc/pom.xml new file mode 100644 index 0000000..9ee3b5e --- /dev/null +++ b/hw03-gc/pom.xml @@ -0,0 +1,38 @@ + + + 4.0.0 + + + ru.otus + otus-java + 2019-09-SNAPSHOT + + + hw03-gc + jar + hw03-gc + + + 11 + 11 + + + + + ch.qos.logback + logback-classic + + + + + ${project.name} + + + org.apache.maven.plugins + maven-assembly-plugin + + + + diff --git a/hw03-gc/src/main/java/ru/otus/homework/GcDemo.java b/hw03-gc/src/main/java/ru/otus/homework/GcDemo.java new file mode 100644 index 0000000..1ddcdab --- /dev/null +++ b/hw03-gc/src/main/java/ru/otus/homework/GcDemo.java @@ -0,0 +1,72 @@ +package ru.otus.homework; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * -Xms512m + * -Xmx512m + * -Xlog:gc=debug:file=./logs/gc.log:tags,uptime,time,level:filecount=10,filesize=10m + * -XX:+HeapDumpOnOutOfMemoryError + * -XX:HeapDumpPath=./logs/dump + * -XX:+UseG1GC + *

+ * -XX:+UseSerialGC + * -XX:+UseParallelGC + * -XX:+UseConcMarkSweepGC + * -XX:+UseG1GC + */ +public class GcDemo { + private static final Logger LOG = LoggerFactory.getLogger(GcDemo.class); + + public static void main(String[] args) { + new GcDemo().startDemo(); + } + + private void startDemo() { + Map> map = new HashMap<>(); + GcStatistic gcStatistic = new GcStatistic(map); + gcStatistic.switchOnMonitoring(); + + long beginTime = System.currentTimeMillis(); + List list = readFile(); + + MemoryLeakApp appDemo = new MemoryLeakApp(list); + try { + appDemo.run(); + } catch (OutOfMemoryError e) { + LOG.info("time: {}", (System.currentTimeMillis() - beginTime) / 1000); + LOG.info("GC statistic: {}", map); + LOG.info("Out of memory error", e); + } + } + + private List readFile() { + List list = new ArrayList<>(); + try { + ClassLoader classLoader = Thread.currentThread().getContextClassLoader(); + InputStream in = classLoader.getResourceAsStream("example.txt"); + try (BufferedReader reader = new BufferedReader(new InputStreamReader(in, StandardCharsets.UTF_8))) { + String line; + while ((line = reader.readLine()) != null) { + String[] ar = line.split(" "); + list.addAll(Arrays.asList(ar)); + } + } + } catch (IOException e) { + LOG.error("Can not read file ", e); + } + return list; + } +} diff --git a/hw03-gc/src/main/java/ru/otus/homework/GcStatistic.java b/hw03-gc/src/main/java/ru/otus/homework/GcStatistic.java new file mode 100644 index 0000000..d780d2d --- /dev/null +++ b/hw03-gc/src/main/java/ru/otus/homework/GcStatistic.java @@ -0,0 +1,62 @@ +package ru.otus.homework; + +import com.sun.management.GarbageCollectionNotificationInfo; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.management.Notification; +import javax.management.NotificationEmitter; +import javax.management.NotificationListener; +import javax.management.openmbean.CompositeData; +import java.lang.management.GarbageCollectorMXBean; +import java.lang.management.ManagementFactory; +import java.lang.management.MemoryMXBean; +import java.util.Collections; +import java.util.List; +import java.util.Map; + +public class GcStatistic { + private static final Logger LOG = LoggerFactory.getLogger(GcStatistic.class); + private final Map> map; + + public GcStatistic(Map> map) { + this.map = map; + } + + public void switchOnMonitoring() { + List gcBeans = ManagementFactory.getGarbageCollectorMXBeans(); + final MemoryMXBean memoryMXBean = ManagementFactory.getMemoryMXBean(); + for (GarbageCollectorMXBean gcbean : gcBeans) { + LOG.info("GC name: {}", gcbean.getName()); + NotificationEmitter emitter = (NotificationEmitter) gcbean; + NotificationListener listener = (Notification notification, Object handback) -> { + if (notification.getType().equals(GarbageCollectionNotificationInfo.GARBAGE_COLLECTION_NOTIFICATION)) { + handleNotification(notification, + memoryMXBean.getHeapMemoryUsage().getUsed()); + } + }; + emitter.addNotificationListener(listener, null, null); + } + } + + private void handleNotification(Notification notification, long memoryUsed) { + GarbageCollectionNotificationInfo info = + GarbageCollectionNotificationInfo.from((CompositeData) notification.getUserData()); + String gcName = info.getGcName(); + long duration = info.getGcInfo().getDuration(); + + String heapMemory = String.format("%.2f Mb", (double)memoryUsed / 1_048_576); + LOG.info("Name: {}, action: {}, cause: {} ({} ms). Used heap memory {}", + info.getGcName(), info.getGcAction(), info.getGcCause(), duration, + heapMemory); + + if (map.containsKey(gcName)) { + Map actionStatistic = map.get(gcName); + long eventCount = actionStatistic.entrySet().iterator().next().getKey(); + long sumDuration = actionStatistic.get(eventCount) + duration; + map.put(gcName, Collections.singletonMap(eventCount + 1, sumDuration)); + } else { + map.put(gcName, Collections.singletonMap(1L, duration)); + } + } +} diff --git a/hw03-gc/src/main/java/ru/otus/homework/MemoryLeakApp.java b/hw03-gc/src/main/java/ru/otus/homework/MemoryLeakApp.java new file mode 100644 index 0000000..7f1a62b --- /dev/null +++ b/hw03-gc/src/main/java/ru/otus/homework/MemoryLeakApp.java @@ -0,0 +1,24 @@ +package ru.otus.homework; + +import java.util.ArrayList; +import java.util.List; + +public class MemoryLeakApp { + private static final List TEMP = new ArrayList<>(); + private static final int ITERATION = 100_000; + private final List init = new ArrayList<>(); + + public MemoryLeakApp(List in) { + init.addAll(in); + } + + public void run() { + for (int i = 0; i < ITERATION; i++) { + TEMP.addAll(init); + for (int j = 0; j < ITERATION / 2; j++) { + TEMP.add("empty" + i); + } + TEMP.removeAll(init); + } + } +} diff --git a/hw03-gc/src/main/resources/example.txt b/hw03-gc/src/main/resources/example.txt new file mode 100644 index 0000000..bd0602d --- /dev/null +++ b/hw03-gc/src/main/resources/example.txt @@ -0,0 +1,7 @@ +start:618 Name:G1 Young Generation, action:end of minor GC, gcCause:G1 Evacuation Pause(88 ms) start:779 Name:G1 Young Generation, action:end of minor GC, gcCause:G1 Evacuation Pause(114 ms) +start:939 Name:G1 Young Generation, action:end of minor GC, gcCause:G1 Evacuation Pause(85 ms) start:1076 Name:G1 Young Generation, action:end of minor GC, gcCause:G1 Evacuation Pause(53 ms) +start:1159 Name:G1 Young Generation, action:end of minor GC, gcCause:G1 Evacuation Pause(52 ms) start:1350 Name:G1 Young Generation, action:end of minor GC, gcCause:G1 Evacuation Pause(44 ms) +start:1414 Name:G1 Young Generation, action:end of minor GC, gcCause:G1 Evacuation Pause(37 ms) start:1476 Name:G1 Young Generation, action:end of minor GC, gcCause:G1 Evacuation Pause(69 ms) +start:1579 Name:G1 Young Generation, action:end of minor GC, gcCause:G1 Evacuation Pause(98 ms) start:1803 Name:G1 Young Generation, action:end of minor GC, gcCause:G1 Humongous Allocation(72 ms) +start:1935 Name:G1 Young Generation, action:end of minor GC, gcCause:G1 Evacuation Pause(154 ms) start:2141 Name:G1 Young Generation, action:end of minor GC, gcCause:G1 Evacuation Pause(108 ms) +start:2411 Name:G1 Young Generation, action:end of minor GC, gcCause:G1 Evacuation Pause(105 ms) start:2614 Name:G1 Young Generation, action:end of minor GC, gcCause:G1 Evacuation Pause(130 ms) diff --git a/hw03-gc/src/main/resources/logback.xml b/hw03-gc/src/main/resources/logback.xml new file mode 100644 index 0000000..08fc1ec --- /dev/null +++ b/hw03-gc/src/main/resources/logback.xml @@ -0,0 +1,13 @@ + + + + + + %d{HH:mm:ss.SSS} - %msg%n + + + + + + + diff --git a/hw04-logging/README.md b/hw04-logging/README.md new file mode 100644 index 0000000..d9dc649 --- /dev/null +++ b/hw04-logging/README.md @@ -0,0 +1,22 @@ +# Автомагическое логирование. + Разработайте такой функционал: + метод класса можно пометить самодельной аннотацией @Log, например, так: + + class TestLogging { + @Log + public void calculation(int param) {}; + } + + При вызове этого метода "автомагически" в консоль должны логироваться + значения параметров. Например так. + + class Demo { + public void action() { + new TestLogging().calculation(6); + } + } + + В консоли дожно быть: + executed method: calculation, param: 6 + + Обратите внимание: явного вызова логирования быть не должно. \ No newline at end of file diff --git a/hw04-logging/pom.xml b/hw04-logging/pom.xml new file mode 100644 index 0000000..8657a35 --- /dev/null +++ b/hw04-logging/pom.xml @@ -0,0 +1,64 @@ + + + 4.0.0 + + + ru.otus + otus-java + 2019-09-SNAPSHOT + + + hw04-logging + hw04-logging + + + 7.2 + 3.2.1 + + + + + org.ow2.asm + asm + ${asm.version} + + + org.ow2.asm + asm-commons + ${asm.version} + + + + + + + maven-shade-plugin + ${maven-shade-plugin.version} + + + asmDemo + package + + shade + + + target/asmDemo.jar + + + + ru.otus.homework.demo.AsmDemo + ru.otus.homework.asm.Agent + + + + + + + + + + + diff --git a/hw04-logging/src/main/java/ru/otus/homework/annotation/Log.java b/hw04-logging/src/main/java/ru/otus/homework/annotation/Log.java new file mode 100644 index 0000000..2b490d0 --- /dev/null +++ b/hw04-logging/src/main/java/ru/otus/homework/annotation/Log.java @@ -0,0 +1,11 @@ +package ru.otus.homework.annotation; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +@Target(ElementType.METHOD) +@Retention(RetentionPolicy.RUNTIME) +public @interface Log { +} diff --git a/hw04-logging/src/main/java/ru/otus/homework/asm/Agent.java b/hw04-logging/src/main/java/ru/otus/homework/asm/Agent.java new file mode 100644 index 0000000..c0c96f8 --- /dev/null +++ b/hw04-logging/src/main/java/ru/otus/homework/asm/Agent.java @@ -0,0 +1,49 @@ +package ru.otus.homework.asm; + +import org.objectweb.asm.ClassReader; +import org.objectweb.asm.ClassVisitor; +import org.objectweb.asm.ClassWriter; +import org.objectweb.asm.Opcodes; + +import java.io.FileOutputStream; +import java.io.IOException; +import java.io.OutputStream; +import java.lang.instrument.ClassFileTransformer; +import java.lang.instrument.Instrumentation; +import java.security.ProtectionDomain; + +public class Agent { + private static final String OUTPUT_FILE = "proxyASM.class"; + + public static void premain(String agentArgs, Instrumentation inst) { + + inst.addTransformer(new ClassFileTransformer() { + @Override + public byte[] transform(ClassLoader loader, String className, + Class classBeingRedefined, + ProtectionDomain protectionDomain, + byte[] classfileBuffer) { + if (className.equals(agentArgs)) { + return scanLogAnnotation(classfileBuffer); + } + return classfileBuffer; + } + }); + } + + private static byte[] scanLogAnnotation(byte[] classfileBuffer) { + ClassReader reader = new ClassReader(classfileBuffer); + ClassWriter writer = new ClassWriter(reader, ClassWriter.COMPUTE_MAXS); + ClassVisitor visitor = new LogClassVisitor(writer); + reader.accept(visitor, Opcodes.ASM5); + + byte[] finalClass = writer.toByteArray(); + + try (OutputStream fos = new FileOutputStream(OUTPUT_FILE)) { + fos.write(finalClass); + } catch (IOException e) { + e.printStackTrace(); + } + return finalClass; + } +} diff --git a/hw04-logging/src/main/java/ru/otus/homework/asm/LogClassVisitor.java b/hw04-logging/src/main/java/ru/otus/homework/asm/LogClassVisitor.java new file mode 100755 index 0000000..12fb5ec --- /dev/null +++ b/hw04-logging/src/main/java/ru/otus/homework/asm/LogClassVisitor.java @@ -0,0 +1,21 @@ +package ru.otus.homework.asm; + +import org.objectweb.asm.ClassVisitor; +import org.objectweb.asm.MethodVisitor; +import org.objectweb.asm.Opcodes; + +public class LogClassVisitor extends ClassVisitor implements Opcodes { + + public LogClassVisitor(ClassVisitor writer) { + super(ASM5, writer); + } + + @Override + public MethodVisitor visitMethod(int access, String methodName, String desc, + String signature, String[] exceptions) { + + MethodVisitor methodVisitor = cv.visitMethod(access, methodName, desc, signature, + exceptions); + return new LogMethodVisitor(methodVisitor, methodName, desc, access); + } +} diff --git a/hw04-logging/src/main/java/ru/otus/homework/asm/LogMethodVisitor.java b/hw04-logging/src/main/java/ru/otus/homework/asm/LogMethodVisitor.java new file mode 100755 index 0000000..8a3398f --- /dev/null +++ b/hw04-logging/src/main/java/ru/otus/homework/asm/LogMethodVisitor.java @@ -0,0 +1,93 @@ +package ru.otus.homework.asm; + +import org.objectweb.asm.AnnotationVisitor; +import org.objectweb.asm.MethodVisitor; +import org.objectweb.asm.Opcodes; +import org.objectweb.asm.Type; + +/** + * Add logging if method has @Log annotation + */ +public class LogMethodVisitor extends MethodVisitor implements Opcodes { + private static final String LOG_ANNOTATION = "Lru/otus/homework/annotation/Log;"; + + private final String methodName; + private final String descriptor; + private final int access; + private boolean isAnnotationPresent; + + public LogMethodVisitor(MethodVisitor methodVisitor, String methodName, String descriptor, int access) { + super(ASM5, methodVisitor); + this.methodName = methodName; + this.descriptor = descriptor; + this.access = access; + } + + @Override + public AnnotationVisitor visitAnnotation(String desc, boolean visible) { + if (LOG_ANNOTATION.equals(desc)) { + isAnnotationPresent = true; + } + return super.visitAnnotation(desc, visible); + } + + @Override + public void visitCode() { + if (isAnnotationPresent) { + Type[] argTypes = Type.getArgumentTypes(descriptor); + int i = (access | ACC_STATIC) == 0 ? 0 : 1; + int index = argTypes.length + i; + + mv.visitIntInsn(BIPUSH, argTypes.length); + mv.visitTypeInsn(ANEWARRAY, "java/lang/Object"); + + int x = 0; + for (Type t : argTypes) { + mv.visitInsn(DUP); + mv.visitIntInsn(BIPUSH, x); + if (t.equals(Type.BOOLEAN_TYPE)) { + mv.visitVarInsn(ILOAD, i); + invokeStatic("java/lang/Boolean", "(Z)Ljava/lang/Boolean;"); + } else if (t.equals(Type.BYTE_TYPE)) { + mv.visitVarInsn(ILOAD, i); + invokeStatic("java/lang/Byte", "(B)Ljava/lang/Byte;"); + } else if (t.equals(Type.CHAR_TYPE)) { + mv.visitVarInsn(ILOAD, i); + invokeStatic("java/lang/Character", "(C)Ljava/lang/Character;"); + } else if (t.equals(Type.SHORT_TYPE)) { + mv.visitVarInsn(ILOAD, i); + invokeStatic("java/lang/Short", "(S)Ljava/lang/Short;"); + } else if (t.equals(Type.INT_TYPE)) { + mv.visitVarInsn(ILOAD, i); + invokeStatic("java/lang/Integer", "(I)Ljava/lang/Integer;"); + } else if (t.equals(Type.LONG_TYPE)) { + mv.visitVarInsn(LLOAD, i); + invokeStatic("java/lang/Long", "(J)Ljava/lang/Long;"); + i++; + } else if (t.equals(Type.FLOAT_TYPE)) { + mv.visitVarInsn(FLOAD, i); + invokeStatic("java/lang/Float", "(F)Ljava/lang/Float;"); + } else if (t.equals(Type.DOUBLE_TYPE)) { + mv.visitVarInsn(DLOAD, i); + invokeStatic("java/lang/Double", "(D)Ljava/lang/Double;"); + i++; + } else { + mv.visitVarInsn(ALOAD, i); + } + mv.visitInsn(AASTORE); + i++; + x++; + } + mv.visitVarInsn(ASTORE, index); + this.visitLdcInsn(methodName); + this.visitVarInsn(ALOAD, index); + mv.visitMethodInsn(INVOKESTATIC, "ru/otus/homework/asm/Logger", "logMethod", + "(Ljava/lang/String;[Ljava/lang/Object;)V", false); + } + super.visitCode(); + } + + private void invokeStatic(String owner, String descriptor) { + mv.visitMethodInsn(INVOKESTATIC, owner, "valueOf", descriptor, false); + } +} diff --git a/hw04-logging/src/main/java/ru/otus/homework/asm/Logger.java b/hw04-logging/src/main/java/ru/otus/homework/asm/Logger.java new file mode 100755 index 0000000..5cfc943 --- /dev/null +++ b/hw04-logging/src/main/java/ru/otus/homework/asm/Logger.java @@ -0,0 +1,22 @@ +package ru.otus.homework.asm; + +import java.util.Arrays; +import java.util.Objects; +import java.util.stream.Collectors; + +/** + * This class is called by LogMethodVisitor to log method`s name and arguments + */ +public class Logger { + + public static void logMethod(String methodName, Object[] args) { + String log = String.format("executed method %s", methodName); + if (args.length > 0) { + String params = Arrays.stream(args). + filter(Objects::nonNull). + map(Object::toString).collect(Collectors.joining(", ")); + log = log.concat(", params: ").concat(params); + } + System.out.println(log); + } +} diff --git a/hw04-logging/src/main/java/ru/otus/homework/demo/AsmDemo.java b/hw04-logging/src/main/java/ru/otus/homework/demo/AsmDemo.java new file mode 100755 index 0000000..eff8dc9 --- /dev/null +++ b/hw04-logging/src/main/java/ru/otus/homework/demo/AsmDemo.java @@ -0,0 +1,21 @@ +package ru.otus.homework.demo; + +/* + * Class with methods which has @Log annotation should be passed as argument for javaagent: + * + * java -javaagent:target/asmDemo.jar=ru/otus/homework/demo/TestLogging -jar target/asmDemo.jar + */ +public class AsmDemo { + + public static void main(String[] args) { + TestLogging testLogging = new TestLogging(); + testLogging.testNoArgs(); + testLogging.testNoLog(); + testLogging.testObject(new TestObject("string", 357)); + testLogging.testInt(3, 2, 1); + testLogging.testIntegerFloatDouble(1, 2.1f, 3.111111); + testLogging.testBooleanShortByte(true, (short) 1, (byte) 1); + testLogging.testLongDouble(5L, 7.8); + testLogging.testStringChar("asdfg", 't'); + } +} diff --git a/hw04-logging/src/main/java/ru/otus/homework/demo/TestLogging.java b/hw04-logging/src/main/java/ru/otus/homework/demo/TestLogging.java new file mode 100755 index 0000000..d4bb803 --- /dev/null +++ b/hw04-logging/src/main/java/ru/otus/homework/demo/TestLogging.java @@ -0,0 +1,49 @@ +package ru.otus.homework.demo; + +import ru.otus.homework.annotation.Log; + +/** + * Test class + */ +public class TestLogging { + + @Log + public void testInt(int param1, int param2, int param3) { + System.out.println("Multiple int"); + } + + @Log + public void testIntegerFloatDouble(Integer i, Float f, Double d) { + System.out.println("Test integer, float"); + } + + @Log + public void testBooleanShortByte(boolean bool, short s, byte b) { + System.out.println("Test boolean, short, byte"); + } + + @Log + public void testLongDouble(long l, double d){ + System.out.println("Test long, double"); + } + + public void testNoLog() { + System.out.println("Log annotation is absent"); + } + + @Log + public void testNoArgs() { + System.out.println("No arguments"); + } + + @Log + public void testObject(TestObject o) { + System.out.println("Test object"); + } + + @Log + public String testStringChar(String s, char a) { + System.out.println("Test string and char"); + return s; + } +} diff --git a/hw04-logging/src/main/java/ru/otus/homework/demo/TestObject.java b/hw04-logging/src/main/java/ru/otus/homework/demo/TestObject.java new file mode 100644 index 0000000..83a1afd --- /dev/null +++ b/hw04-logging/src/main/java/ru/otus/homework/demo/TestObject.java @@ -0,0 +1,19 @@ +package ru.otus.homework.demo; + +public class TestObject { + private String s; + private int i; + + public TestObject(String s, int i) { + this.s = s; + this.i = i; + } + + @Override + public String toString() { + return "TestObject{" + + "s='" + s + '\'' + + ", i=" + i + + '}'; + } +} diff --git a/hw15-messageSystem/README.md b/hw15-messageSystem/README.md new file mode 100644 index 0000000..da1b4c6 --- /dev/null +++ b/hw15-messageSystem/README.md @@ -0,0 +1,3 @@ +# MessageSystem + Добавить систему обмена сообщениями в ДЗ про веб сервер с IoC контейнером. + Пересылать сообщения из вебсокета в DBService и обратно. \ No newline at end of file diff --git a/hw15-messageSystem/pom.xml b/hw15-messageSystem/pom.xml new file mode 100644 index 0000000..f093616 --- /dev/null +++ b/hw15-messageSystem/pom.xml @@ -0,0 +1,78 @@ + + + 4.0.0 + + + org.springframework.boot + spring-boot-starter-parent + 2.2.2.RELEASE + + + hw15-messageSystem + hw15-messageSystem + + + + org.springframework.boot + spring-boot-starter-web + + + org.springframework.boot + spring-boot-starter-thymeleaf + + + org.springframework.boot + spring-boot-starter-data-mongodb + + + + org.springframework.boot + spring-boot-starter-websocket + + + + org.webjars + stomp-websocket + 2.3.3-1 + + + org.webjars + webjars-locator-core + + + org.webjars + sockjs-client + 1.1.2 + + + org.webjars + bootstrap + 4.3.1 + + + org.webjars + jquery + 3.1.0 + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + org.apache.maven.plugins + maven-compiler-plugin + + 11 + 11 + + + + + + diff --git a/hw15-messageSystem/src/main/java/ru/otus/homework/Application.java b/hw15-messageSystem/src/main/java/ru/otus/homework/Application.java new file mode 100644 index 0000000..67c91e5 --- /dev/null +++ b/hw15-messageSystem/src/main/java/ru/otus/homework/Application.java @@ -0,0 +1,12 @@ +package ru.otus.homework; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; + +@SpringBootApplication +public class Application { + + public static void main(String[] args) { + SpringApplication.run(Application.class); + } +} diff --git a/hw15-messageSystem/src/main/java/ru/otus/homework/cache/CacheListener.java b/hw15-messageSystem/src/main/java/ru/otus/homework/cache/CacheListener.java new file mode 100644 index 0000000..b8f793f --- /dev/null +++ b/hw15-messageSystem/src/main/java/ru/otus/homework/cache/CacheListener.java @@ -0,0 +1,5 @@ +package ru.otus.homework.cache; + +public interface CacheListener { + void notify(K key, V value, String action); +} diff --git a/hw15-messageSystem/src/main/java/ru/otus/homework/cache/CacheService.java b/hw15-messageSystem/src/main/java/ru/otus/homework/cache/CacheService.java new file mode 100644 index 0000000..2664415 --- /dev/null +++ b/hw15-messageSystem/src/main/java/ru/otus/homework/cache/CacheService.java @@ -0,0 +1,14 @@ +package ru.otus.homework.cache; + +public interface CacheService { + + void put(K key, V value); + + void remove(K key); + + V get(K key); + + void addListener(CacheListener listener); + + void removeListener(CacheListener listener); +} diff --git a/hw15-messageSystem/src/main/java/ru/otus/homework/cache/CacheServiceImpl.java b/hw15-messageSystem/src/main/java/ru/otus/homework/cache/CacheServiceImpl.java new file mode 100644 index 0000000..c002be7 --- /dev/null +++ b/hw15-messageSystem/src/main/java/ru/otus/homework/cache/CacheServiceImpl.java @@ -0,0 +1,73 @@ +package ru.otus.homework.cache; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Service; + +import java.lang.ref.Reference; +import java.lang.ref.ReferenceQueue; +import java.lang.ref.WeakReference; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.WeakHashMap; + +@Service +public class CacheServiceImpl implements CacheService { + private static final Logger LOG = LoggerFactory.getLogger(CacheServiceImpl.class); + private final Map cache = new WeakHashMap<>(); + private final List>> listeners = new ArrayList<>(); + private final ReferenceQueue> refQueue = new ReferenceQueue<>(); + + @Override + public void put(K key, V value) { + notifyListener(key, value, "add"); + cache.put(key, value); + } + + @Override + public void remove(K key) { + notifyListener(key, cache.get(key), "remove"); + cache.remove(key); + } + + @Override + public V get(K key) { + return cache.get(key); + } + + @Override + public void addListener(CacheListener listener) { + WeakReference> weakReference = new WeakReference<>(listener, refQueue); + listeners.add(weakReference); + } + + @Override + public void removeListener(CacheListener listener) { + Optional>> reference = listeners.stream(). + filter(ref -> listener.equals(ref.get())).findAny(); + reference.ifPresent(listeners::remove); + } + + private void notifyListener(K key, V value, String action) { + checkListeners(); + listeners.forEach((WeakReference> listener) -> { + if (listener != null) { + try { + Objects.requireNonNull(listener.get()).notify(key, value, action); + } catch (Exception e) { + LOG.error("Error in listener: {}", e.getMessage()); + } + } + }); + } + + private void checkListeners() { + Reference> reference = refQueue.poll(); + if (reference != null) { + listeners.remove(reference); + } + } +} diff --git a/hw15-messageSystem/src/main/java/ru/otus/homework/common/Serializers.java b/hw15-messageSystem/src/main/java/ru/otus/homework/common/Serializers.java new file mode 100644 index 0000000..c25088c --- /dev/null +++ b/hw15-messageSystem/src/main/java/ru/otus/homework/common/Serializers.java @@ -0,0 +1,40 @@ +package ru.otus.homework.common; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import ru.otus.homework.exception.UserSerializationException; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; + +public final class Serializers { + private static final Logger LOG = LoggerFactory.getLogger(Serializers.class); + + private Serializers() { + } + + public static byte[] serialize(Object data) { + try (ByteArrayOutputStream baos = new ByteArrayOutputStream(); + ObjectOutputStream os = new ObjectOutputStream(baos)) { + os.writeObject(data); + os.flush(); + return baos.toByteArray(); + } catch (Exception e) { + LOG.error("Serialization error, data: {}", data); + throw new UserSerializationException("Serialization error", e.getCause()); + } + } + + public static T deserialize(byte[] data, Class classOfT) { + try (ByteArrayInputStream bis = new ByteArrayInputStream(data); + ObjectInputStream is = new ObjectInputStream(bis)) { + Object obj = is.readObject(); + return classOfT.cast(obj); + } catch (Exception e) { + LOG.error("DeSerialization error, classOfT: {}", classOfT.getName()); + throw new UserSerializationException("DeSerialization error", e.getCause()); + } + } +} diff --git a/hw15-messageSystem/src/main/java/ru/otus/homework/config/MessageSystemConfig.java b/hw15-messageSystem/src/main/java/ru/otus/homework/config/MessageSystemConfig.java new file mode 100644 index 0000000..5aba5e3 --- /dev/null +++ b/hw15-messageSystem/src/main/java/ru/otus/homework/config/MessageSystemConfig.java @@ -0,0 +1,67 @@ +package ru.otus.homework.config; + +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.event.ContextRefreshedEvent; +import org.springframework.context.event.EventListener; +import ru.otus.homework.front.FrontendService; +import ru.otus.homework.front.FrontendServiceImpl; +import ru.otus.homework.front.handlers.UserDataResponseHandler; +import ru.otus.homework.messagesystem.MessageSystem; +import ru.otus.homework.messagesystem.MessageType; +import ru.otus.homework.messagesystem.MsClient; +import ru.otus.homework.messagesystem.MsClientImpl; +import ru.otus.homework.messagesystem.RequestHandler; +import ru.otus.homework.service.UserService; +import ru.otus.homework.service.handlers.UserDataRequestHandler; + +@Configuration +public class MessageSystemConfig { + private static final String FRONTEND_CLIENT_NAME = "frontendService"; + private static final String DATABASE_CLIENT_NAME = "databaseService"; + + private final MessageSystem messageSystem; + private final UserService userService; + + public MessageSystemConfig(MessageSystem messageSystem, UserService userService) { + this.messageSystem = messageSystem; + this.userService = userService; + } + + @Bean + public RequestHandler requestHandler() { + return new UserDataRequestHandler(userService); + } + + @Bean + public MsClient databaseClient() { + MsClient msClient = new MsClientImpl(DATABASE_CLIENT_NAME, messageSystem); + msClient.addHandler(MessageType.USER_SAVE, requestHandler()); + msClient.addHandler(MessageType.USER_LIST, requestHandler()); + return msClient; + } + + @Bean + MsClient frontendClient() { + return new MsClientImpl(FRONTEND_CLIENT_NAME, messageSystem); + } + + @Bean + public FrontendService frontendService() { + return new FrontendServiceImpl(frontendClient(), DATABASE_CLIENT_NAME); + } + + @Bean + public RequestHandler responseHandler() { + return new UserDataResponseHandler(frontendService()); + } + + @EventListener + public void onApplicationEvent(ContextRefreshedEvent event) { + frontendClient().addHandler(MessageType.USER_SAVE, responseHandler()); + frontendClient().addHandler(MessageType.USER_LIST, responseHandler()); + + messageSystem.addClient(frontendClient()); + messageSystem.addClient(databaseClient()); + } +} diff --git a/hw15-messageSystem/src/main/java/ru/otus/homework/config/WebSocketConfig.java b/hw15-messageSystem/src/main/java/ru/otus/homework/config/WebSocketConfig.java new file mode 100644 index 0000000..015b88f --- /dev/null +++ b/hw15-messageSystem/src/main/java/ru/otus/homework/config/WebSocketConfig.java @@ -0,0 +1,23 @@ +package ru.otus.homework.config; + +import org.springframework.context.annotation.Configuration; +import org.springframework.messaging.simp.config.MessageBrokerRegistry; +import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker; +import org.springframework.web.socket.config.annotation.StompEndpointRegistry; +import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer; + +@Configuration +@EnableWebSocketMessageBroker +public class WebSocketConfig implements WebSocketMessageBrokerConfigurer { + + @Override + public void configureMessageBroker(MessageBrokerRegistry config) { + config.enableSimpleBroker("/topic"); + config.setApplicationDestinationPrefixes("/app"); + } + + @Override + public void registerStompEndpoints(StompEndpointRegistry registry) { + registry.addEndpoint("/ws").withSockJS(); + } +} diff --git a/hw15-messageSystem/src/main/java/ru/otus/homework/controller/UserController.java b/hw15-messageSystem/src/main/java/ru/otus/homework/controller/UserController.java new file mode 100644 index 0000000..d128ddc --- /dev/null +++ b/hw15-messageSystem/src/main/java/ru/otus/homework/controller/UserController.java @@ -0,0 +1,56 @@ +package ru.otus.homework.controller; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.messaging.handler.annotation.MessageMapping; +import org.springframework.messaging.handler.annotation.Payload; +import org.springframework.messaging.simp.SimpMessagingTemplate; +import org.springframework.stereotype.Controller; +import org.springframework.web.bind.annotation.GetMapping; +import ru.otus.homework.dto.UserDto; +import ru.otus.homework.front.FrontendService; + +import java.util.List; + +@Controller +public class UserController { + private static final Logger LOG = LoggerFactory.getLogger(UserController.class); + + private final FrontendService frontendService; + private final SimpMessagingTemplate template; + + public UserController(FrontendService frontendService, SimpMessagingTemplate template) { + this.frontendService = frontendService; + this.template = template; + } + + @GetMapping("/user/save") + public String userSaveView() { + return "userSave"; + } + + @MessageMapping("/user/save") + public void saveUser(@Payload UserDto userDto) { + LOG.info("save user {}", userDto); + + frontendService.saveUser(userDto, (String message) -> { + LOG.info("response data:{}", message); + template.convertAndSend("/topic/response/user/save", message); + }); + } + + @GetMapping("/user/list") + public String userListView() { + return "userList"; + } + + @MessageMapping("/user/list") + public void userList() { + LOG.info("get list of users"); + + frontendService.getUserList((List data) -> { + LOG.info("response data:{}", data); + template.convertAndSend("/topic/response/user/list", data); + }); + } +} diff --git a/hw15-messageSystem/src/main/java/ru/otus/homework/dto/UserDto.java b/hw15-messageSystem/src/main/java/ru/otus/homework/dto/UserDto.java new file mode 100644 index 0000000..3f5736b --- /dev/null +++ b/hw15-messageSystem/src/main/java/ru/otus/homework/dto/UserDto.java @@ -0,0 +1,61 @@ +package ru.otus.homework.dto; + +import ru.otus.homework.model.Role; + +import java.io.Serializable; + +public class UserDto implements Serializable { + private String name; + private String login; + private String password; + private Role role; + + public UserDto(String name, String login, String password, Role role) { + this.name = name; + this.login = login; + this.password = password; + this.role = role; + } + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + public String getLogin() { + return login; + } + + public void setLogin(String login) { + this.login = login; + } + + public String getPassword() { + return password; + } + + public void setPassword(String password) { + this.password = password; + } + + public Role getRole() { + return role; + } + + public void setRole(Role role) { + this.role = role; + } + + @Override + public String toString() { + return "UserDto{" + + "name='" + name + '\'' + + ", login='" + login + '\'' + + ", password='" + password + '\'' + + ", role=" + role + + '}'; + } +} diff --git a/hw15-messageSystem/src/main/java/ru/otus/homework/exception/DataHandlerException.java b/hw15-messageSystem/src/main/java/ru/otus/homework/exception/DataHandlerException.java new file mode 100644 index 0000000..b4df6df --- /dev/null +++ b/hw15-messageSystem/src/main/java/ru/otus/homework/exception/DataHandlerException.java @@ -0,0 +1,8 @@ +package ru.otus.homework.exception; + +public class DataHandlerException extends RuntimeException { + + public DataHandlerException(String message) { + super(message); + } +} diff --git a/hw15-messageSystem/src/main/java/ru/otus/homework/exception/UserSerializationException.java b/hw15-messageSystem/src/main/java/ru/otus/homework/exception/UserSerializationException.java new file mode 100644 index 0000000..3164295 --- /dev/null +++ b/hw15-messageSystem/src/main/java/ru/otus/homework/exception/UserSerializationException.java @@ -0,0 +1,8 @@ +package ru.otus.homework.exception; + +public class UserSerializationException extends RuntimeException { + + public UserSerializationException(String error, Throwable cause) { + super(error, cause); + } +} diff --git a/hw15-messageSystem/src/main/java/ru/otus/homework/exception/UserValidationException.java b/hw15-messageSystem/src/main/java/ru/otus/homework/exception/UserValidationException.java new file mode 100644 index 0000000..5a05a0c --- /dev/null +++ b/hw15-messageSystem/src/main/java/ru/otus/homework/exception/UserValidationException.java @@ -0,0 +1,12 @@ +package ru.otus.homework.exception; + +public class UserValidationException extends RuntimeException { + + public UserValidationException(String message) { + super(message); + } + + public UserValidationException(String message, Throwable cause) { + super(message, cause); + } +} diff --git a/hw15-messageSystem/src/main/java/ru/otus/homework/front/FrontendService.java b/hw15-messageSystem/src/main/java/ru/otus/homework/front/FrontendService.java new file mode 100644 index 0000000..d5a2f6b --- /dev/null +++ b/hw15-messageSystem/src/main/java/ru/otus/homework/front/FrontendService.java @@ -0,0 +1,17 @@ +package ru.otus.homework.front; + +import ru.otus.homework.dto.UserDto; + +import java.util.List; +import java.util.Optional; +import java.util.UUID; +import java.util.function.Consumer; + +public interface FrontendService { + + void saveUser(UserDto userDto, Consumer dataConsumer); + + void getUserList(Consumer> dataConsumer); + + Optional> takeConsumer(UUID sourceMessageId, Class tClass); +} diff --git a/hw15-messageSystem/src/main/java/ru/otus/homework/front/FrontendServiceImpl.java b/hw15-messageSystem/src/main/java/ru/otus/homework/front/FrontendServiceImpl.java new file mode 100644 index 0000000..f27855b --- /dev/null +++ b/hw15-messageSystem/src/main/java/ru/otus/homework/front/FrontendServiceImpl.java @@ -0,0 +1,52 @@ +package ru.otus.homework.front; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import ru.otus.homework.dto.UserDto; +import ru.otus.homework.messagesystem.Message; +import ru.otus.homework.messagesystem.MessageType; +import ru.otus.homework.messagesystem.MsClient; + +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Consumer; + +public class FrontendServiceImpl implements FrontendService { + private static final Logger LOG = LoggerFactory.getLogger(FrontendServiceImpl.class); + private final Map> consumerMap = new ConcurrentHashMap<>(); + + private final MsClient msClient; + private final String databaseServiceClientName; + + public FrontendServiceImpl(MsClient msClient, String databaseServiceClientName) { + this.msClient = msClient; + this.databaseServiceClientName = databaseServiceClientName; + } + + @Override + public void saveUser(UserDto userDto, Consumer dataConsumer) { + Message outMsg = msClient.produceMessage(databaseServiceClientName, userDto, MessageType.USER_SAVE); + consumerMap.put(outMsg.getId(), dataConsumer); + msClient.sendMessage(outMsg); + } + + @Override + public void getUserList(Consumer> dataConsumer) { + Message outMsg = msClient.produceMessage(databaseServiceClientName, null, MessageType.USER_LIST); + consumerMap.put(outMsg.getId(), dataConsumer); + msClient.sendMessage(outMsg); + } + + @Override + public Optional> takeConsumer(UUID sourceMessageId, Class tClass) { + Consumer consumer = (Consumer) consumerMap.remove(sourceMessageId); + if (consumer == null) { + LOG.warn("consumer not found for:{}", sourceMessageId); + return Optional.empty(); + } + return Optional.of(consumer); + } +} diff --git a/hw15-messageSystem/src/main/java/ru/otus/homework/front/handlers/UserDataResponseHandler.java b/hw15-messageSystem/src/main/java/ru/otus/homework/front/handlers/UserDataResponseHandler.java new file mode 100644 index 0000000..f0737b8 --- /dev/null +++ b/hw15-messageSystem/src/main/java/ru/otus/homework/front/handlers/UserDataResponseHandler.java @@ -0,0 +1,50 @@ +package ru.otus.homework.front.handlers; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import ru.otus.homework.common.Serializers; +import ru.otus.homework.dto.UserDto; +import ru.otus.homework.exception.DataHandlerException; +import ru.otus.homework.front.FrontendService; +import ru.otus.homework.messagesystem.Message; +import ru.otus.homework.messagesystem.RequestHandler; + +import java.util.List; +import java.util.Optional; +import java.util.UUID; + +public class UserDataResponseHandler implements RequestHandler { + private static final Logger LOG = LoggerFactory.getLogger(UserDataResponseHandler.class); + private final FrontendService frontendService; + + public UserDataResponseHandler(FrontendService frontendService) { + this.frontendService = frontendService; + } + + @Override + public Optional handle(Message msg) { + LOG.info("new message:{}", msg); + try { + UUID sourceMessageId = msg.getSourceMessageId(). + orElseThrow(() -> new RuntimeException("Not found sourceMsg for message:" + msg.getId())); + String messageType = msg.getType(); + switch (messageType) { + case "UserSave": + String message = Serializers.deserialize(msg.getPayload(), String.class); + frontendService.takeConsumer(sourceMessageId, String.class). + ifPresent(consumer -> consumer.accept(message)); + break; + case "UserList": + List users = Serializers.deserialize(msg.getPayload(), List.class); + frontendService.takeConsumer(sourceMessageId, List.class). + ifPresent(consumer -> consumer.accept(users)); + break; + default: + throw new DataHandlerException(String.format("Unknown message type %s", messageType)); + } + } catch (Exception ex) { + LOG.error("msg: {}", ex.getMessage()); + } + return Optional.empty(); + } +} diff --git a/hw15-messageSystem/src/main/java/ru/otus/homework/loaddata/DataLoader.java b/hw15-messageSystem/src/main/java/ru/otus/homework/loaddata/DataLoader.java new file mode 100644 index 0000000..e3c7c28 --- /dev/null +++ b/hw15-messageSystem/src/main/java/ru/otus/homework/loaddata/DataLoader.java @@ -0,0 +1,28 @@ +package ru.otus.homework.loaddata; + +import org.springframework.context.ApplicationListener; +import org.springframework.context.event.ContextRefreshedEvent; +import org.springframework.stereotype.Component; +import ru.otus.homework.model.Role; +import ru.otus.homework.model.AppUser; +import ru.otus.homework.repository.UserRepository; + +/** + * Create default user in data base + */ +@Component +public class DataLoader implements ApplicationListener { + private static final String ADMIN = "admin"; + private final UserRepository userRepository; + + public DataLoader(UserRepository userRepository) { + this.userRepository = userRepository; + } + + @Override + public void onApplicationEvent(final ContextRefreshedEvent event) { + if (userRepository.getByLogin(ADMIN).isEmpty()) { + userRepository.save(new AppUser(ADMIN, ADMIN, ADMIN, Role.ADMIN)); + } + } +} diff --git a/hw15-messageSystem/src/main/java/ru/otus/homework/messagesystem/Message.java b/hw15-messageSystem/src/main/java/ru/otus/homework/messagesystem/Message.java new file mode 100644 index 0000000..0a62fe4 --- /dev/null +++ b/hw15-messageSystem/src/main/java/ru/otus/homework/messagesystem/Message.java @@ -0,0 +1,88 @@ +package ru.otus.homework.messagesystem; + +import java.util.Arrays; +import java.util.Objects; +import java.util.Optional; +import java.util.UUID; + +public class Message { + static final Message VOID_MESSAGE = new Message(); + private final UUID id = UUID.randomUUID(); + private final String from; + private final String to; + private final UUID sourceMessageId; + private final String type; + private int payloadLength; + private final byte[] payload; + + private Message() { + this.from = null; + this.to = null; + this.sourceMessageId = null; + this.type = "voidTechnicalMessage"; + this.payload = new byte[1]; + this.payloadLength = payload.length; + } + + public Message(String from, String to, UUID sourceMessageId, String type, byte[] payload) { + this.from = from; + this.to = to; + this.sourceMessageId = sourceMessageId; + this.type = type; + this.payloadLength = payload.length; + this.payload = Arrays.copyOf(payload, payloadLength); + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + Message message = (Message) o; + return id == message.id; + } + + @Override + public int hashCode() { + return Objects.hash(id); + } + + @Override + public String toString() { + return "Message{" + + "id=" + id + + ", from='" + from + '\'' + + ", to='" + to + '\'' + + ", sourceMessageId=" + sourceMessageId + + ", type='" + type + '\'' + + ", payloadLength=" + payloadLength + + '}'; + } + + public UUID getId() { + return id; + } + + public String getFrom() { + return from; + } + + public String getTo() { + return to; + } + + public String getType() { + return type; + } + + public byte[] getPayload() { + return Arrays.copyOf(payload, payloadLength); + } + + public int getPayloadLength() { + return payloadLength; + } + + public Optional getSourceMessageId() { + return Optional.ofNullable(sourceMessageId); + } +} diff --git a/hw15-messageSystem/src/main/java/ru/otus/homework/messagesystem/MessageSystem.java b/hw15-messageSystem/src/main/java/ru/otus/homework/messagesystem/MessageSystem.java new file mode 100644 index 0000000..a480dcd --- /dev/null +++ b/hw15-messageSystem/src/main/java/ru/otus/homework/messagesystem/MessageSystem.java @@ -0,0 +1,19 @@ +package ru.otus.homework.messagesystem; + +public interface MessageSystem { + + void addClient(MsClient msClient); + + void removeClient(String clientId); + + boolean newMessage(Message msg); + + void dispose() throws InterruptedException; + + void dispose(Runnable callback) throws InterruptedException; + + void start(); + + int currentQueueSize(); +} + diff --git a/hw15-messageSystem/src/main/java/ru/otus/homework/messagesystem/MessageSystemImpl.java b/hw15-messageSystem/src/main/java/ru/otus/homework/messagesystem/MessageSystemImpl.java new file mode 100644 index 0000000..493c319 --- /dev/null +++ b/hw15-messageSystem/src/main/java/ru/otus/homework/messagesystem/MessageSystemImpl.java @@ -0,0 +1,164 @@ +package ru.otus.homework.messagesystem; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Service; + +import java.util.Map; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +@Service +public final class MessageSystemImpl implements MessageSystem { + private static final Logger LOG = LoggerFactory.getLogger(MessageSystemImpl.class); + private static final int MESSAGE_QUEUE_SIZE = 100_000; + private static final int MSG_HANDLER_THREAD_LIMIT = 2; + + private final AtomicBoolean runFlag = new AtomicBoolean(true); + + private final Map clientMap = new ConcurrentHashMap<>(); + private final BlockingQueue messageQueue = new ArrayBlockingQueue<>(MESSAGE_QUEUE_SIZE); + + private Runnable disposeCallback; + + private final ExecutorService msgProcessor = Executors.newSingleThreadExecutor(runnable -> { + Thread thread = new Thread(runnable); + thread.setName("msg-processor-thread"); + return thread; + }); + + private final ExecutorService msgHandler = + Executors.newFixedThreadPool(MSG_HANDLER_THREAD_LIMIT, new ThreadFactory() { + private final AtomicInteger threadNameSeq = new AtomicInteger(0); + + @Override + public Thread newThread(Runnable runnable) { + Thread thread = new Thread(runnable); + thread.setName("msg-handler-thread-" + threadNameSeq.incrementAndGet()); + return thread; + } + }); + + public MessageSystemImpl() { + start(); + } + + public MessageSystemImpl(boolean startProcessing) { + if (startProcessing) { + start(); + } + } + + @Override + public void start() { + msgProcessor.submit(this::msgProcessor); + } + + @Override + public int currentQueueSize() { + return messageQueue.size(); + } + + @Override + public void addClient(MsClient msClient) { + LOG.info("new client:{}", msClient.getName()); + if (clientMap.containsKey(msClient.getName())) { + throw new IllegalArgumentException("Error. client: " + msClient.getName() + " already exists"); + } + clientMap.put(msClient.getName(), msClient); + } + + @Override + public void removeClient(String clientId) { + MsClient removedClient = clientMap.remove(clientId); + if (removedClient == null) { + LOG.warn("client not found: {}", clientId); + } else { + LOG.info("removed client:{}", removedClient); + } + } + + @Override + public boolean newMessage(Message msg) { + if (runFlag.get()) { + return messageQueue.offer(msg); + } else { + LOG.warn("MS is being shutting down... rejected:{}", msg); + return false; + } + } + + @Override + public void dispose() throws InterruptedException { + LOG.info("now in the messageQueue {} messages", currentQueueSize()); + runFlag.set(false); + insertStopMessage(); + msgProcessor.shutdown(); + msgHandler.awaitTermination(60, TimeUnit.SECONDS); + } + + @Override + public void dispose(Runnable callback) throws InterruptedException { + disposeCallback = callback; + dispose(); + } + + private void msgProcessor() { + LOG.info("msgProcessor started, {}", currentQueueSize()); + while (runFlag.get() || !messageQueue.isEmpty()) { + try { + Message msg = messageQueue.take(); + if (msg == Message.VOID_MESSAGE) { + LOG.info("received the stop message"); + } else { + MsClient clientTo = clientMap.get(msg.getTo()); + if (clientTo == null) { + LOG.warn("client not found"); + } else { + msgHandler.submit(() -> handleMessage(clientTo, msg)); + } + } + } catch (InterruptedException ex) { + LOG.error(ex.getMessage(), ex); + Thread.currentThread().interrupt(); + } catch (Exception ex) { + LOG.error(ex.getMessage(), ex); + } + } + + if (disposeCallback != null) { + msgHandler.submit(disposeCallback); + } + msgHandler.submit(this::messageHandlerShutdown); + LOG.info("msgProcessor finished"); + } + + private void messageHandlerShutdown() { + msgHandler.shutdown(); + LOG.info("msgHandler has been shut down"); + } + + private void handleMessage(MsClient msClient, Message msg) { + try { + msClient.handle(msg); + } catch (Exception ex) { + LOG.error(ex.getMessage(), ex); + LOG.error("message:{}", msg); + } + } + + private void insertStopMessage() throws InterruptedException { + boolean result = messageQueue.offer(Message.VOID_MESSAGE); + while (!result) { + Thread.sleep(100); + result = messageQueue.offer(Message.VOID_MESSAGE); + } + } +} diff --git a/hw15-messageSystem/src/main/java/ru/otus/homework/messagesystem/MessageType.java b/hw15-messageSystem/src/main/java/ru/otus/homework/messagesystem/MessageType.java new file mode 100644 index 0000000..4c89596 --- /dev/null +++ b/hw15-messageSystem/src/main/java/ru/otus/homework/messagesystem/MessageType.java @@ -0,0 +1,16 @@ +package ru.otus.homework.messagesystem; + +public enum MessageType { + USER_SAVE("UserSave"), + USER_LIST("UserList"); + + private final String value; + + MessageType(String value) { + this.value = value; + } + + public String getValue() { + return value; + } +} diff --git a/hw15-messageSystem/src/main/java/ru/otus/homework/messagesystem/MsClient.java b/hw15-messageSystem/src/main/java/ru/otus/homework/messagesystem/MsClient.java new file mode 100644 index 0000000..b084d65 --- /dev/null +++ b/hw15-messageSystem/src/main/java/ru/otus/homework/messagesystem/MsClient.java @@ -0,0 +1,14 @@ +package ru.otus.homework.messagesystem; + +public interface MsClient { + + void addHandler(MessageType type, RequestHandler requestHandler); + + boolean sendMessage(Message msg); + + void handle(Message msg); + + String getName(); + + Message produceMessage(String to, T data, MessageType msgType); +} diff --git a/hw15-messageSystem/src/main/java/ru/otus/homework/messagesystem/MsClientImpl.java b/hw15-messageSystem/src/main/java/ru/otus/homework/messagesystem/MsClientImpl.java new file mode 100644 index 0000000..752e9a8 --- /dev/null +++ b/hw15-messageSystem/src/main/java/ru/otus/homework/messagesystem/MsClientImpl.java @@ -0,0 +1,73 @@ +package ru.otus.homework.messagesystem; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import ru.otus.homework.common.Serializers; + +import java.util.Map; +import java.util.Objects; +import java.util.concurrent.ConcurrentHashMap; + +public class MsClientImpl implements MsClient { + private static final Logger LOG = LoggerFactory.getLogger(MsClientImpl.class); + private final String name; + private final MessageSystem messageSystem; + private final Map handlers = new ConcurrentHashMap<>(); + + public MsClientImpl(String name, MessageSystem messageSystem) { + this.name = name; + this.messageSystem = messageSystem; + } + + @Override + public void addHandler(MessageType type, RequestHandler requestHandler) { + this.handlers.put(type.getValue(), requestHandler); + } + + @Override + public String getName() { + return name; + } + + @Override + public boolean sendMessage(Message msg) { + boolean result = messageSystem.newMessage(msg); + if (!result) { + LOG.error("the last message was rejected: {}", msg); + } + return result; + } + + @Override + public void handle(Message msg) { + LOG.info("new message:{}", msg); + try { + RequestHandler requestHandler = handlers.get(msg.getType()); + if (requestHandler != null) { + requestHandler.handle(msg).ifPresent(this::sendMessage); + } else { + LOG.error("handler not found for the message type:{}", msg.getType()); + } + } catch (Exception ex) { + LOG.error("msg: {}", msg, ex); + } + } + + @Override + public Message produceMessage(String to, T data, MessageType msgType) { + return new Message(name, to, null, msgType.getValue(), Serializers.serialize(data)); + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + MsClientImpl msClient = (MsClientImpl) o; + return Objects.equals(name, msClient.name); + } + + @Override + public int hashCode() { + return Objects.hash(name); + } +} diff --git a/hw15-messageSystem/src/main/java/ru/otus/homework/messagesystem/RequestHandler.java b/hw15-messageSystem/src/main/java/ru/otus/homework/messagesystem/RequestHandler.java new file mode 100644 index 0000000..d33272a --- /dev/null +++ b/hw15-messageSystem/src/main/java/ru/otus/homework/messagesystem/RequestHandler.java @@ -0,0 +1,7 @@ +package ru.otus.homework.messagesystem; + +import java.util.Optional; + +public interface RequestHandler { + Optional handle(Message msg); +} diff --git a/hw15-messageSystem/src/main/java/ru/otus/homework/model/AppUser.java b/hw15-messageSystem/src/main/java/ru/otus/homework/model/AppUser.java new file mode 100644 index 0000000..9570466 --- /dev/null +++ b/hw15-messageSystem/src/main/java/ru/otus/homework/model/AppUser.java @@ -0,0 +1,84 @@ +package ru.otus.homework.model; + +import org.springframework.data.mongodb.core.index.Indexed; +import org.springframework.data.mongodb.core.mapping.Document; + +import java.util.Objects; + +@Document(collection = "users") +public class AppUser { + private String name; + @Indexed(unique = true) + private String login; + private String password; + private Role role; + + public AppUser(String name, String login, String password, Role role) { + this.name = name; + this.login = login; + this.password = password; + this.role = role; + } + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + public String getLogin() { + return login; + } + + public void setLogin(String login) { + this.login = login; + } + + public String getPassword() { + return password; + } + + public void setPassword(String password) { + this.password = password; + } + + public Role getRole() { + return role; + } + + public void setRole(Role role) { + this.role = role; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + AppUser user = (AppUser) o; + return Objects.equals(name, user.name) && + Objects.equals(login, user.login) && + Objects.equals(password, user.password) && + role == user.role; + } + + @Override + public int hashCode() { + return Objects.hash(name, login, password, role); + } + + @Override + public String toString() { + return "User{" + + "name='" + name + '\'' + + ", login='" + login + '\'' + + ", password='" + password + '\'' + + ", role=" + role + + '}'; + } +} diff --git a/hw15-messageSystem/src/main/java/ru/otus/homework/model/Role.java b/hw15-messageSystem/src/main/java/ru/otus/homework/model/Role.java new file mode 100644 index 0000000..64f6a8a --- /dev/null +++ b/hw15-messageSystem/src/main/java/ru/otus/homework/model/Role.java @@ -0,0 +1,16 @@ +package ru.otus.homework.model; + +public enum Role { + ADMIN("admin"), + USER("user"); + + private final String displayName; + + Role(String displayName) { + this.displayName = displayName; + } + + public String getDisplayName() { + return displayName; + } +} diff --git a/hw15-messageSystem/src/main/java/ru/otus/homework/repository/UserRepository.java b/hw15-messageSystem/src/main/java/ru/otus/homework/repository/UserRepository.java new file mode 100644 index 0000000..eaed5d3 --- /dev/null +++ b/hw15-messageSystem/src/main/java/ru/otus/homework/repository/UserRepository.java @@ -0,0 +1,11 @@ +package ru.otus.homework.repository; + +import org.springframework.data.mongodb.repository.MongoRepository; +import ru.otus.homework.model.AppUser; + +import java.util.Optional; + +public interface UserRepository extends MongoRepository { + + Optional getByLogin(String login); +} diff --git a/hw15-messageSystem/src/main/java/ru/otus/homework/service/UserService.java b/hw15-messageSystem/src/main/java/ru/otus/homework/service/UserService.java new file mode 100644 index 0000000..5ea9364 --- /dev/null +++ b/hw15-messageSystem/src/main/java/ru/otus/homework/service/UserService.java @@ -0,0 +1,12 @@ +package ru.otus.homework.service; + +import ru.otus.homework.model.AppUser; + +import java.util.List; + +public interface UserService { + + AppUser saveUser(AppUser user); + + List getUsers(); +} diff --git a/hw15-messageSystem/src/main/java/ru/otus/homework/service/UserServiceImpl.java b/hw15-messageSystem/src/main/java/ru/otus/homework/service/UserServiceImpl.java new file mode 100644 index 0000000..2a5e5b4 --- /dev/null +++ b/hw15-messageSystem/src/main/java/ru/otus/homework/service/UserServiceImpl.java @@ -0,0 +1,48 @@ +package ru.otus.homework.service; + +import org.springframework.dao.DuplicateKeyException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Service; +import ru.otus.homework.model.AppUser; +import ru.otus.homework.repository.UserRepository; +import ru.otus.homework.cache.CacheService; +import ru.otus.homework.exception.UserValidationException; + +import java.util.List; + +@Service +public class UserServiceImpl implements UserService { + private static final Logger LOG = LoggerFactory.getLogger(UserServiceImpl.class); + + private final UserRepository userRepository; + private final CacheService cacheService; + + public UserServiceImpl(UserRepository userRepository, CacheService cacheService) { + this.userRepository = userRepository; + this.cacheService = cacheService; + cacheService.addListener((key, value, action) -> LoggerFactory.getLogger( + "CacheListener").info("key:{}, value:{}, action: {}", key, value, action)); + } + + @Override + public AppUser saveUser(AppUser user) { + LOG.info("save user: {}", user.getLogin()); + if (cacheService.get(user.getLogin()) != null) { + throw new UserValidationException("Login already exists"); + } + try { + AppUser savedUser = userRepository.save(user); + cacheService.put(savedUser.getLogin(), savedUser); + return savedUser; + } catch (DuplicateKeyException e) { + throw new UserValidationException("Login already exists", e); + } + } + + @Override + public List getUsers() { + LOG.info("get users"); + return userRepository.findAll(); + } +} diff --git a/hw15-messageSystem/src/main/java/ru/otus/homework/service/handlers/UserDataRequestHandler.java b/hw15-messageSystem/src/main/java/ru/otus/homework/service/handlers/UserDataRequestHandler.java new file mode 100644 index 0000000..e7d1a94 --- /dev/null +++ b/hw15-messageSystem/src/main/java/ru/otus/homework/service/handlers/UserDataRequestHandler.java @@ -0,0 +1,67 @@ +package ru.otus.homework.service.handlers; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import ru.otus.homework.common.Serializers; +import ru.otus.homework.dto.UserDto; +import ru.otus.homework.exception.DataHandlerException; +import ru.otus.homework.exception.UserValidationException; +import ru.otus.homework.messagesystem.Message; +import ru.otus.homework.messagesystem.RequestHandler; +import ru.otus.homework.model.AppUser; +import ru.otus.homework.service.UserService; + +import java.util.List; +import java.util.Optional; +import java.util.stream.Collectors; + +public class UserDataRequestHandler implements RequestHandler { + private static final Logger LOG = LoggerFactory.getLogger(UserDataRequestHandler.class); + private final UserService userService; + + public UserDataRequestHandler(UserService userService) { + this.userService = userService; + } + + @Override + public Optional handle(Message msg) { + LOG.info("handle msg: {}", msg); + + String messageType = msg.getType(); + Message returnMessage; + switch (messageType) { + case "UserSave": + UserDto user = Serializers.deserialize(msg.getPayload(), UserDto.class); + returnMessage = new Message(msg.getTo(), msg.getFrom(), msg.getId(), + msg.getType(), Serializers.serialize(getMessage(user))); + break; + case "UserList": + List responseList = userService.getUsers().stream(). + map(this::toDto).collect(Collectors.toList()); + returnMessage = new Message(msg.getTo(), msg.getFrom(), msg.getId(), + msg.getType(), Serializers.serialize(responseList)); + break; + default: + throw new DataHandlerException(String.format("Unknown message type %s", messageType)); + } + return Optional.of(returnMessage); + } + + private String getMessage(UserDto userDto) { + try { + UserDto user = toDto(userService.saveUser(toEntity(userDto))); + return "Saved " + user.getLogin(); + } catch (UserValidationException e) { + LOG.error("UserDataRequestHandler ", e); + return "Error " + e.getMessage(); + } + } + + private AppUser toEntity(UserDto userDto) { + return new AppUser(userDto.getName(), userDto.getLogin(), userDto.getPassword(), userDto.getRole()); + } + + private UserDto toDto(AppUser user) { + return new UserDto(user.getName(), user.getLogin(), user.getPassword(), user.getRole()); + } +} diff --git a/hw15-messageSystem/src/main/resources/application.yml b/hw15-messageSystem/src/main/resources/application.yml new file mode 100644 index 0000000..5fb54dc --- /dev/null +++ b/hw15-messageSystem/src/main/resources/application.yml @@ -0,0 +1,12 @@ +spring: + data: + mongodb: + host: localhost + port: 27017 + database: testDI + mvc: + throw-exception-if-no-handler-found: true + +logging.level: + ROOT: INFO + org.springframework: INFO diff --git a/hw15-messageSystem/src/main/resources/logback.xml b/hw15-messageSystem/src/main/resources/logback.xml new file mode 100644 index 0000000..31883bd --- /dev/null +++ b/hw15-messageSystem/src/main/resources/logback.xml @@ -0,0 +1,13 @@ + + + + + %d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n + + + + + + + + diff --git a/hw15-messageSystem/src/main/resources/static/list.js b/hw15-messageSystem/src/main/resources/static/list.js new file mode 100644 index 0000000..8756091 --- /dev/null +++ b/hw15-messageSystem/src/main/resources/static/list.js @@ -0,0 +1,27 @@ +var stompClient = null; + +function getUserList() { + stompClient = Stomp.over(new SockJS('/ws')); + + stompClient.connect({}, function () { + console.log('connected stompClient'); + stompClient.send("/app/user/list", {}, ''); + stompClient.subscribe('/topic/response/user/list', function (output) { + console.log('response from /user/list'); + var usersResp = JSON.parse(output.body); + + var tableData = '' + for (i in usersResp) { + tableData+= ''; + tableData+= '' + usersResp[i].name + ''; + tableData+= '' + usersResp[i].login + ''; + tableData+= '' + usersResp[i].role + ''; + tableData+= ''; + } + tableData+=''; + document.getElementById('tableData').innerHTML = tableData; + }); + }, function (err) { + alert('error' + err); + }); +} diff --git a/hw15-messageSystem/src/main/resources/static/save.js b/hw15-messageSystem/src/main/resources/static/save.js new file mode 100644 index 0000000..6816835 --- /dev/null +++ b/hw15-messageSystem/src/main/resources/static/save.js @@ -0,0 +1,30 @@ +var stompClient = null; + +function connect() { + stompClient = Stomp.over(new SockJS('/ws')); + stompClient.connect({}, function () { + console.log('connected stompClient'); + stompClient.subscribe('/topic/response/user/save', function (output) { + console.log('response from user/save'); + alert(output.body); + }); + }, function (err) { + alert('error' + err); + }); +} + +function saveUser() { + var name = $("#name").val(); + var login = $("#login").val(); + var password = $("#password").val(); + var role = $("#role").val(); + + if(!login || 0 === login.length || !password || 0 === password.length){ + alert('Login and password can not be null'); + } + stompClient.send("/app/user/save", {}, JSON.stringify({name, login, password, role})); +}; + +$(() => { + $("form").on('submit', event => event.preventDefault()); +}); diff --git a/hw15-messageSystem/src/main/resources/templates/error.html b/hw15-messageSystem/src/main/resources/templates/error.html new file mode 100644 index 0000000..3a2d75e --- /dev/null +++ b/hw15-messageSystem/src/main/resources/templates/error.html @@ -0,0 +1,13 @@ + + + + Error + + +

+

Error

+

+ List of all users +
+ + diff --git a/hw15-messageSystem/src/main/resources/templates/userList.html b/hw15-messageSystem/src/main/resources/templates/userList.html new file mode 100644 index 0000000..1040b19 --- /dev/null +++ b/hw15-messageSystem/src/main/resources/templates/userList.html @@ -0,0 +1,38 @@ + + + + + + Users + + + + + + + + + + + +
+

Users

+ + + + + + + + + +
NameLoginRole
+
+

+
+
+ + diff --git a/hw15-messageSystem/src/main/resources/templates/userSave.html b/hw15-messageSystem/src/main/resources/templates/userSave.html new file mode 100644 index 0000000..2b4b245 --- /dev/null +++ b/hw15-messageSystem/src/main/resources/templates/userSave.html @@ -0,0 +1,61 @@ + + + + + + Admin page + + + + + + + + + + + +
+

Save new user:

+
+
+
+
+
+ + +
+
+ + +
+
+ + +
+
+ +
+ +
+
+
+
+ +
+
+
+

+ List of all users +
+
+
+ + diff --git a/hw16_common/pom.xml b/hw16_common/pom.xml new file mode 100644 index 0000000..bd068c9 --- /dev/null +++ b/hw16_common/pom.xml @@ -0,0 +1,36 @@ + + + 4.0.0 + + otus-java + ru.otus + 2019-09-SNAPSHOT + + + hw16_common + + + UTF-8 + 11 + 11 + + + + + ch.qos.logback + logback-classic + + + + + ${project.name} + + + org.apache.maven.plugins + maven-assembly-plugin + + + + diff --git a/hw16_common/src/main/java/ru/otus/homework/common/Serializers.java b/hw16_common/src/main/java/ru/otus/homework/common/Serializers.java new file mode 100644 index 0000000..2de0a42 --- /dev/null +++ b/hw16_common/src/main/java/ru/otus/homework/common/Serializers.java @@ -0,0 +1,40 @@ +package ru.otus.homework.common; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import ru.otus.homework.exception.UserSerializationException; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; + +public final class Serializers { + private static final Logger LOG = LoggerFactory.getLogger(Serializers.class); + + private Serializers() { + } + + public static byte[] serialize(Object data) { + try (ByteArrayOutputStream baos = new ByteArrayOutputStream(); + ObjectOutputStream os = new ObjectOutputStream(baos)) { + os.writeObject(data); + os.flush(); + return baos.toByteArray(); + } catch (Exception e) { + LOG.error("Serialization error, data: {}", data); + throw new UserSerializationException("Serialization error", e); + } + } + + public static T deserialize(byte[] data, Class classOfT) { + try (ByteArrayInputStream bis = new ByteArrayInputStream(data); + ObjectInputStream is = new ObjectInputStream(bis)) { + Object obj = is.readObject(); + return classOfT.cast(obj); + } catch (Exception e) { + LOG.error("DeSerialization error, classOfT: {}", classOfT.getName()); + throw new UserSerializationException("DeSerialization error", e); + } + } +} diff --git a/hw16_common/src/main/java/ru/otus/homework/dto/Role.java b/hw16_common/src/main/java/ru/otus/homework/dto/Role.java new file mode 100644 index 0000000..7848758 --- /dev/null +++ b/hw16_common/src/main/java/ru/otus/homework/dto/Role.java @@ -0,0 +1,16 @@ +package ru.otus.homework.dto; + +public enum Role { + ADMIN("admin"), + USER("user"); + + private final String displayName; + + Role(String displayName) { + this.displayName = displayName; + } + + public String getDisplayName() { + return displayName; + } +} diff --git a/hw16_common/src/main/java/ru/otus/homework/dto/UserDto.java b/hw16_common/src/main/java/ru/otus/homework/dto/UserDto.java new file mode 100644 index 0000000..77a6dcf --- /dev/null +++ b/hw16_common/src/main/java/ru/otus/homework/dto/UserDto.java @@ -0,0 +1,62 @@ +package ru.otus.homework.dto; + +import java.io.Serializable; + +public class UserDto implements Serializable { + private String name; + private String login; + private String password; + private Role role; + + public UserDto() { + } + + public UserDto(String name, String login, String password, Role role) { + this.name = name; + this.login = login; + this.password = password; + this.role = role; + } + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + public String getLogin() { + return login; + } + + public void setLogin(String login) { + this.login = login; + } + + public String getPassword() { + return password; + } + + public void setPassword(String password) { + this.password = password; + } + + public Role getRole() { + return role; + } + + public void setRole(Role role) { + this.role = role; + } + + @Override + public String toString() { + return "UserDto{" + + "name='" + name + '\'' + + ", login='" + login + '\'' + + ", password='" + password + '\'' + + ", role=" + role + + '}'; + } +} diff --git a/hw16_common/src/main/java/ru/otus/homework/exception/DataHandlerException.java b/hw16_common/src/main/java/ru/otus/homework/exception/DataHandlerException.java new file mode 100644 index 0000000..e28cfdd --- /dev/null +++ b/hw16_common/src/main/java/ru/otus/homework/exception/DataHandlerException.java @@ -0,0 +1,12 @@ +package ru.otus.homework.exception; + +public class DataHandlerException extends RuntimeException { + + public DataHandlerException(String message) { + super(message); + } + + public DataHandlerException(String message, Exception e) { + super(message, e); + } +} diff --git a/hw16_common/src/main/java/ru/otus/homework/exception/UserSerializationException.java b/hw16_common/src/main/java/ru/otus/homework/exception/UserSerializationException.java new file mode 100644 index 0000000..3164295 --- /dev/null +++ b/hw16_common/src/main/java/ru/otus/homework/exception/UserSerializationException.java @@ -0,0 +1,8 @@ +package ru.otus.homework.exception; + +public class UserSerializationException extends RuntimeException { + + public UserSerializationException(String error, Throwable cause) { + super(error, cause); + } +} diff --git a/hw16_common/src/main/java/ru/otus/homework/exception/UserValidationException.java b/hw16_common/src/main/java/ru/otus/homework/exception/UserValidationException.java new file mode 100644 index 0000000..5a05a0c --- /dev/null +++ b/hw16_common/src/main/java/ru/otus/homework/exception/UserValidationException.java @@ -0,0 +1,12 @@ +package ru.otus.homework.exception; + +public class UserValidationException extends RuntimeException { + + public UserValidationException(String message) { + super(message); + } + + public UserValidationException(String message, Throwable cause) { + super(message, cause); + } +} diff --git a/hw16_common/src/main/java/ru/otus/homework/ms/Message.java b/hw16_common/src/main/java/ru/otus/homework/ms/Message.java new file mode 100644 index 0000000..3fa9316 --- /dev/null +++ b/hw16_common/src/main/java/ru/otus/homework/ms/Message.java @@ -0,0 +1,89 @@ +package ru.otus.homework.ms; + +import java.io.Serializable; +import java.util.Arrays; +import java.util.Objects; +import java.util.Optional; +import java.util.UUID; + +public class Message implements Serializable { + public static final Message VOID_MESSAGE = new Message(); + private final UUID id = UUID.randomUUID(); + private final String from; + private final String to; + private final UUID sourceMessageId; + private final String type; + private int payloadLength; + private final byte[] payload; + + private Message() { + this.from = null; + this.to = null; + this.sourceMessageId = null; + this.type = "voidTechnicalMessage"; + this.payload = new byte[1]; + this.payloadLength = payload.length; + } + + public Message(String from, String to, UUID sourceMessageId, String type, byte[] payload) { + this.from = from; + this.to = to; + this.sourceMessageId = sourceMessageId; + this.type = type; + this.payloadLength = payload.length; + this.payload = Arrays.copyOf(payload, payloadLength); + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + Message message = (Message) o; + return id == message.id; + } + + @Override + public int hashCode() { + return Objects.hash(id); + } + + @Override + public String toString() { + return "Message{" + + "id=" + id + + ", from='" + from + '\'' + + ", to='" + to + '\'' + + ", sourceMessageId=" + sourceMessageId + + ", type='" + type + '\'' + + ", payloadLength=" + payloadLength + + '}'; + } + + public UUID getId() { + return id; + } + + public String getFrom() { + return from; + } + + public String getTo() { + return to; + } + + public String getType() { + return type; + } + + public byte[] getPayload() { + return Arrays.copyOf(payload, payloadLength); + } + + public int getPayloadLength() { + return payloadLength; + } + + public Optional getSourceMessageId() { + return Optional.ofNullable(sourceMessageId); + } +} diff --git a/hw16_common/src/main/java/ru/otus/homework/ms/MessageType.java b/hw16_common/src/main/java/ru/otus/homework/ms/MessageType.java new file mode 100644 index 0000000..de045ee --- /dev/null +++ b/hw16_common/src/main/java/ru/otus/homework/ms/MessageType.java @@ -0,0 +1,16 @@ +package ru.otus.homework.ms; + +public enum MessageType { + USER_SAVE("UserSave"), + USER_LIST("UserList"); + + private final String value; + + MessageType(String value) { + this.value = value; + } + + public String getValue() { + return value; + } +} diff --git a/hw16_common/src/main/java/ru/otus/homework/ms/MsClient.java b/hw16_common/src/main/java/ru/otus/homework/ms/MsClient.java new file mode 100644 index 0000000..ca111ea --- /dev/null +++ b/hw16_common/src/main/java/ru/otus/homework/ms/MsClient.java @@ -0,0 +1,14 @@ +package ru.otus.homework.ms; + +public interface MsClient { + + void addHandler(MessageType type, RequestHandler requestHandler); + + boolean sendMessage(Message msg); + + void handle(Message msg); + + String getName(); + + Message produceMessage(String to, T data, MessageType msgType); +} diff --git a/hw16_common/src/main/java/ru/otus/homework/ms/MsClientImpl.java b/hw16_common/src/main/java/ru/otus/homework/ms/MsClientImpl.java new file mode 100644 index 0000000..cf2db29 --- /dev/null +++ b/hw16_common/src/main/java/ru/otus/homework/ms/MsClientImpl.java @@ -0,0 +1,83 @@ +package ru.otus.homework.ms; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import ru.otus.homework.common.Serializers; + +import java.io.ObjectOutputStream; +import java.net.Socket; +import java.util.Map; +import java.util.Objects; +import java.util.concurrent.ConcurrentHashMap; + +public class MsClientImpl implements MsClient { + private static final Logger LOG = LoggerFactory.getLogger(MsClientImpl.class); + private final String name; + private final String host; + private final int port; + private final Map handlers = new ConcurrentHashMap<>(); + + public MsClientImpl(String name, String host, int port) { + this.name = name; + this.host = host; + this.port = port; + } + + @Override + public void addHandler(MessageType type, RequestHandler requestHandler) { + this.handlers.put(type.getValue(), requestHandler); + } + + @Override + public String getName() { + return name; + } + + @Override + public boolean sendMessage(Message msg) { + LOG.info("send message {} ", msg); + LOG.info("send to {} {}", host, port); + try (Socket clientSocket = new Socket(host, port)) { + ObjectOutputStream out = new ObjectOutputStream(clientSocket.getOutputStream()); + out.writeObject(msg); + out.writeObject("\r\n"); + } catch (Exception e) { + LOG.error("error, message can not be sent", e); + return false; + } + return true; + } + + @Override + public void handle(Message msg) { + LOG.info("handle message:{}", msg); + try { + RequestHandler requestHandler = handlers.get(msg.getType()); + if (requestHandler != null) { + requestHandler.handle(msg).ifPresent(this::sendMessage); + } else { + LOG.error("handler not found for the message type:{}", msg.getType()); + } + } catch (Exception ex) { + LOG.error("msg: {}", msg, ex); + } + } + + @Override + public Message produceMessage(String to, T data, MessageType msgType) { + return new Message(name, to, null, msgType.getValue(), Serializers.serialize(data)); + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + MsClientImpl msClient = (MsClientImpl) o; + return Objects.equals(name, msClient.name); + } + + @Override + public int hashCode() { + return Objects.hash(name); + } +} diff --git a/hw16_common/src/main/java/ru/otus/homework/ms/RequestHandler.java b/hw16_common/src/main/java/ru/otus/homework/ms/RequestHandler.java new file mode 100644 index 0000000..096ad56 --- /dev/null +++ b/hw16_common/src/main/java/ru/otus/homework/ms/RequestHandler.java @@ -0,0 +1,7 @@ +package ru.otus.homework.ms; + +import java.util.Optional; + +public interface RequestHandler { + Optional handle(Message msg); +} diff --git a/hw16_common/src/main/java/ru/otus/homework/socket/ListenServer.java b/hw16_common/src/main/java/ru/otus/homework/socket/ListenServer.java new file mode 100644 index 0000000..1ff7f7a --- /dev/null +++ b/hw16_common/src/main/java/ru/otus/homework/socket/ListenServer.java @@ -0,0 +1,5 @@ +package ru.otus.homework.socket; + +public interface ListenServer { + void start(); +} diff --git a/hw16_common/src/main/java/ru/otus/homework/socket/ListenServerImpl.java b/hw16_common/src/main/java/ru/otus/homework/socket/ListenServerImpl.java new file mode 100644 index 0000000..6939463 --- /dev/null +++ b/hw16_common/src/main/java/ru/otus/homework/socket/ListenServerImpl.java @@ -0,0 +1,48 @@ +package ru.otus.homework.socket; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import ru.otus.homework.ms.Message; + +import java.io.ObjectInputStream; +import java.net.ServerSocket; +import java.net.Socket; +import java.util.function.Consumer; + +public class ListenServerImpl implements ListenServer { + private static final Logger LOG = LoggerFactory.getLogger(ListenServerImpl.class); + + private final int port; + private final Consumer consumer; + + public ListenServerImpl(int port, Consumer consumer) { + this.port = port; + this.consumer = consumer; + } + + @Override + public void start() { + try (ServerSocket serverSocket = new ServerSocket(port)) { + while (!Thread.currentThread().isInterrupted()) { + LOG.info("waiting for client connection"); + try (Socket clientSocket = serverSocket.accept()) { + clientHandler(clientSocket); + } + } + } catch (Exception ex) { + LOG.error("error", ex); + } + } + + private void clientHandler(Socket clientSocket) { + try (ObjectInputStream in = new ObjectInputStream(clientSocket.getInputStream())) { + Message msg = (Message) in.readObject(); + if (msg != null) { + LOG.info("message from client: {}", msg); + consumer.accept(msg); + } + } catch (Exception ex) { + LOG.error("error", ex); + } + } +} diff --git a/hw16_dbServer/pom.xml b/hw16_dbServer/pom.xml new file mode 100644 index 0000000..044b831 --- /dev/null +++ b/hw16_dbServer/pom.xml @@ -0,0 +1,51 @@ + + + 4.0.0 + + + org.springframework.boot + spring-boot-starter-parent + 2.2.2.RELEASE + + + hw16_dbServer + ru.otus + 2019-09-SNAPSHOT + + + + org.springframework.boot + spring-boot-starter + + + org.springframework.boot + spring-boot-starter-data-mongodb + + + + hw16_common + ru.otus + 2019-09-SNAPSHOT + + + + + dbService + + + org.springframework.boot + spring-boot-maven-plugin + + + org.apache.maven.plugins + maven-compiler-plugin + + 11 + 11 + + + + + diff --git a/hw16_dbServer/src/main/java/ru/otus/homework/DbApplication.java b/hw16_dbServer/src/main/java/ru/otus/homework/DbApplication.java new file mode 100644 index 0000000..d0a0ff5 --- /dev/null +++ b/hw16_dbServer/src/main/java/ru/otus/homework/DbApplication.java @@ -0,0 +1,12 @@ +package ru.otus.homework; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; + +@SpringBootApplication +public class DbApplication { + + public static void main(String[] args) { + SpringApplication.run(DbApplication.class, args); + } +} diff --git a/hw16_dbServer/src/main/java/ru/otus/homework/cache/CacheListener.java b/hw16_dbServer/src/main/java/ru/otus/homework/cache/CacheListener.java new file mode 100644 index 0000000..b8f793f --- /dev/null +++ b/hw16_dbServer/src/main/java/ru/otus/homework/cache/CacheListener.java @@ -0,0 +1,5 @@ +package ru.otus.homework.cache; + +public interface CacheListener { + void notify(K key, V value, String action); +} diff --git a/hw16_dbServer/src/main/java/ru/otus/homework/cache/CacheService.java b/hw16_dbServer/src/main/java/ru/otus/homework/cache/CacheService.java new file mode 100644 index 0000000..2664415 --- /dev/null +++ b/hw16_dbServer/src/main/java/ru/otus/homework/cache/CacheService.java @@ -0,0 +1,14 @@ +package ru.otus.homework.cache; + +public interface CacheService { + + void put(K key, V value); + + void remove(K key); + + V get(K key); + + void addListener(CacheListener listener); + + void removeListener(CacheListener listener); +} diff --git a/hw16_dbServer/src/main/java/ru/otus/homework/cache/CacheServiceImpl.java b/hw16_dbServer/src/main/java/ru/otus/homework/cache/CacheServiceImpl.java new file mode 100644 index 0000000..c002be7 --- /dev/null +++ b/hw16_dbServer/src/main/java/ru/otus/homework/cache/CacheServiceImpl.java @@ -0,0 +1,73 @@ +package ru.otus.homework.cache; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Service; + +import java.lang.ref.Reference; +import java.lang.ref.ReferenceQueue; +import java.lang.ref.WeakReference; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.WeakHashMap; + +@Service +public class CacheServiceImpl implements CacheService { + private static final Logger LOG = LoggerFactory.getLogger(CacheServiceImpl.class); + private final Map cache = new WeakHashMap<>(); + private final List>> listeners = new ArrayList<>(); + private final ReferenceQueue> refQueue = new ReferenceQueue<>(); + + @Override + public void put(K key, V value) { + notifyListener(key, value, "add"); + cache.put(key, value); + } + + @Override + public void remove(K key) { + notifyListener(key, cache.get(key), "remove"); + cache.remove(key); + } + + @Override + public V get(K key) { + return cache.get(key); + } + + @Override + public void addListener(CacheListener listener) { + WeakReference> weakReference = new WeakReference<>(listener, refQueue); + listeners.add(weakReference); + } + + @Override + public void removeListener(CacheListener listener) { + Optional>> reference = listeners.stream(). + filter(ref -> listener.equals(ref.get())).findAny(); + reference.ifPresent(listeners::remove); + } + + private void notifyListener(K key, V value, String action) { + checkListeners(); + listeners.forEach((WeakReference> listener) -> { + if (listener != null) { + try { + Objects.requireNonNull(listener.get()).notify(key, value, action); + } catch (Exception e) { + LOG.error("Error in listener: {}", e.getMessage()); + } + } + }); + } + + private void checkListeners() { + Reference> reference = refQueue.poll(); + if (reference != null) { + listeners.remove(reference); + } + } +} diff --git a/hw16_dbServer/src/main/java/ru/otus/homework/config/DatabaseConfig.java b/hw16_dbServer/src/main/java/ru/otus/homework/config/DatabaseConfig.java new file mode 100644 index 0000000..c551582 --- /dev/null +++ b/hw16_dbServer/src/main/java/ru/otus/homework/config/DatabaseConfig.java @@ -0,0 +1,59 @@ +package ru.otus.homework.config; + +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.event.ContextRefreshedEvent; +import org.springframework.context.event.EventListener; +import ru.otus.homework.handler.UserDataRequestHandler; +import ru.otus.homework.ms.MessageType; +import ru.otus.homework.ms.MsClient; +import ru.otus.homework.ms.MsClientImpl; +import ru.otus.homework.ms.RequestHandler; +import ru.otus.homework.service.UserService; +import ru.otus.homework.socket.ListenServer; +import ru.otus.homework.socket.ListenServerImpl; + +@Configuration +public class DatabaseConfig { + @Value("${ms.client.name}") + private String clientName; + + @Value("${ms.host}") + private String msHost; + + @Value("${ms.port}") + private Integer msPort; + + @Value("${server.port}") + private Integer serverPort; + + private final UserService userService; + + public DatabaseConfig(UserService userService) { + this.userService = userService; + } + + @Bean + public RequestHandler requestHandler() { + return new UserDataRequestHandler(userService); + } + + @Bean + public MsClient databaseClient() { + MsClient msClient = new MsClientImpl(clientName, msHost, msPort); + msClient.addHandler(MessageType.USER_SAVE, requestHandler()); + msClient.addHandler(MessageType.USER_LIST, requestHandler()); + return msClient; + } + + @Bean + public ListenServer databaseServer() { + return new ListenServerImpl(serverPort, msg -> databaseClient().handle(msg)); + } + + @EventListener + public void onApplicationEvent(ContextRefreshedEvent event) { + databaseServer().start(); + } +} diff --git a/hw16_dbServer/src/main/java/ru/otus/homework/handler/UserDataRequestHandler.java b/hw16_dbServer/src/main/java/ru/otus/homework/handler/UserDataRequestHandler.java new file mode 100644 index 0000000..c406bbc --- /dev/null +++ b/hw16_dbServer/src/main/java/ru/otus/homework/handler/UserDataRequestHandler.java @@ -0,0 +1,57 @@ +package ru.otus.homework.handler; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import ru.otus.homework.common.Serializers; +import ru.otus.homework.dto.UserDto; +import ru.otus.homework.exception.DataHandlerException; +import ru.otus.homework.exception.UserValidationException; + +import ru.otus.homework.ms.Message; +import ru.otus.homework.ms.RequestHandler; +import ru.otus.homework.service.UserService; + +import java.util.List; +import java.util.Optional; + +public class UserDataRequestHandler implements RequestHandler { + private static final Logger LOG = LoggerFactory.getLogger(UserDataRequestHandler.class); + private final UserService userService; + + public UserDataRequestHandler(UserService userService) { + this.userService = userService; + } + + @Override + public Optional handle(Message msg) { + LOG.info("handle msg: {}", msg); + + String messageType = msg.getType(); + Message returnMessage; + switch (messageType) { + case "UserSave": + UserDto user = Serializers.deserialize(msg.getPayload(), UserDto.class); + returnMessage = new Message(msg.getTo(), msg.getFrom(), msg.getId(), + msg.getType(), Serializers.serialize(getMessage(user))); + break; + case "UserList": + List responseList = userService.getUsers(); + returnMessage = new Message(msg.getTo(), msg.getFrom(), msg.getId(), + msg.getType(), Serializers.serialize(responseList)); + break; + default: + throw new DataHandlerException(String.format("Unknown message type %s", messageType)); + } + return Optional.of(returnMessage); + } + + private String getMessage(UserDto userDto) { + try { + UserDto savedUser = userService.saveUser(userDto); + return "Saved " + savedUser.getLogin(); + } catch (UserValidationException e) { + LOG.error("UserDataRequestHandler ", e); + return "Error " + e.getMessage(); + } + } +} diff --git a/hw16_dbServer/src/main/java/ru/otus/homework/model/AppUser.java b/hw16_dbServer/src/main/java/ru/otus/homework/model/AppUser.java new file mode 100644 index 0000000..9e77b36 --- /dev/null +++ b/hw16_dbServer/src/main/java/ru/otus/homework/model/AppUser.java @@ -0,0 +1,85 @@ +package ru.otus.homework.model; + +import org.springframework.data.mongodb.core.index.Indexed; +import org.springframework.data.mongodb.core.mapping.Document; +import ru.otus.homework.dto.Role; + +import java.util.Objects; + +@Document(collection = "users") +public class AppUser { + private String name; + @Indexed(unique = true) + private String login; + private String password; + private Role role; + + public AppUser(String name, String login, String password, Role role) { + this.name = name; + this.login = login; + this.password = password; + this.role = role; + } + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + public String getLogin() { + return login; + } + + public void setLogin(String login) { + this.login = login; + } + + public String getPassword() { + return password; + } + + public void setPassword(String password) { + this.password = password; + } + + public Role getRole() { + return role; + } + + public void setRole(Role role) { + this.role = role; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + AppUser user = (AppUser) o; + return Objects.equals(name, user.name) && + Objects.equals(login, user.login) && + Objects.equals(password, user.password) && + role == user.role; + } + + @Override + public int hashCode() { + return Objects.hash(name, login, password, role); + } + + @Override + public String toString() { + return "User{" + + "name='" + name + '\'' + + ", login='" + login + '\'' + + ", password='" + password + '\'' + + ", role=" + role + + '}'; + } +} diff --git a/hw16_dbServer/src/main/java/ru/otus/homework/repository/UserRepository.java b/hw16_dbServer/src/main/java/ru/otus/homework/repository/UserRepository.java new file mode 100644 index 0000000..eaed5d3 --- /dev/null +++ b/hw16_dbServer/src/main/java/ru/otus/homework/repository/UserRepository.java @@ -0,0 +1,11 @@ +package ru.otus.homework.repository; + +import org.springframework.data.mongodb.repository.MongoRepository; +import ru.otus.homework.model.AppUser; + +import java.util.Optional; + +public interface UserRepository extends MongoRepository { + + Optional getByLogin(String login); +} diff --git a/hw16_dbServer/src/main/java/ru/otus/homework/service/UserService.java b/hw16_dbServer/src/main/java/ru/otus/homework/service/UserService.java new file mode 100644 index 0000000..f9fedfc --- /dev/null +++ b/hw16_dbServer/src/main/java/ru/otus/homework/service/UserService.java @@ -0,0 +1,12 @@ +package ru.otus.homework.service; + +import ru.otus.homework.dto.UserDto; + +import java.util.List; + +public interface UserService { + + UserDto saveUser(UserDto user); + + List getUsers(); +} diff --git a/hw16_dbServer/src/main/java/ru/otus/homework/service/UserServiceImpl.java b/hw16_dbServer/src/main/java/ru/otus/homework/service/UserServiceImpl.java new file mode 100644 index 0000000..9a8f87d --- /dev/null +++ b/hw16_dbServer/src/main/java/ru/otus/homework/service/UserServiceImpl.java @@ -0,0 +1,58 @@ +package ru.otus.homework.service; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.dao.DuplicateKeyException; +import org.springframework.stereotype.Service; +import ru.otus.homework.cache.CacheService; +import ru.otus.homework.dto.UserDto; +import ru.otus.homework.exception.UserValidationException; +import ru.otus.homework.model.AppUser; +import ru.otus.homework.repository.UserRepository; + +import java.util.List; +import java.util.stream.Collectors; + +@Service +public class UserServiceImpl implements UserService { + private static final Logger LOG = LoggerFactory.getLogger(UserServiceImpl.class); + + private final UserRepository userRepository; + private final CacheService cacheService; + + public UserServiceImpl(UserRepository userRepository, CacheService cacheService) { + this.userRepository = userRepository; + this.cacheService = cacheService; + cacheService.addListener((key, value, action) -> LoggerFactory.getLogger( + "CacheListener").info("key:{}, value:{}, action: {}", key, value, action)); + } + + @Override + public UserDto saveUser(UserDto user) { + LOG.info("save user: {}", user.getLogin()); + if (cacheService.get(user.getLogin()) != null) { + throw new UserValidationException("Login already exists"); + } + try { + AppUser savedUser = userRepository.save(toEntity(user)); + cacheService.put(savedUser.getLogin(), savedUser); + return toDto(savedUser); + } catch (DuplicateKeyException e) { + throw new UserValidationException("Login already exists", e); + } + } + + @Override + public List getUsers() { + LOG.info("get users"); + return userRepository.findAll().stream().map(this::toDto).collect(Collectors.toList()); + } + + private AppUser toEntity(UserDto userDto) { + return new AppUser(userDto.getName(), userDto.getLogin(), userDto.getPassword(), userDto.getRole()); + } + + private UserDto toDto(AppUser user) { + return new UserDto(user.getName(), user.getLogin(), user.getPassword(), user.getRole()); + } +} diff --git a/hw16_dbServer/src/main/resources/application-db2.yml b/hw16_dbServer/src/main/resources/application-db2.yml new file mode 100644 index 0000000..9651237 --- /dev/null +++ b/hw16_dbServer/src/main/resources/application-db2.yml @@ -0,0 +1,6 @@ +server: + port: 8084 + +ms: + client: + name: database2 diff --git a/hw16_dbServer/src/main/resources/application.yml b/hw16_dbServer/src/main/resources/application.yml new file mode 100644 index 0000000..40a08af --- /dev/null +++ b/hw16_dbServer/src/main/resources/application.yml @@ -0,0 +1,19 @@ +server: + port: 8083 + +ms: + host: localhost + port: 8081 + client: + name: database1 + +spring: + data: + mongodb: + host: localhost + port: 27017 + database: testDI + +logging.level: + ROOT: INFO + org.springframework: INFO diff --git a/hw16_dbServer/src/main/resources/logback.xml b/hw16_dbServer/src/main/resources/logback.xml new file mode 100644 index 0000000..31883bd --- /dev/null +++ b/hw16_dbServer/src/main/resources/logback.xml @@ -0,0 +1,13 @@ + + + + + %d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n + + + + + + + + diff --git a/hw16_frontend/pom.xml b/hw16_frontend/pom.xml new file mode 100644 index 0000000..7e96f9e --- /dev/null +++ b/hw16_frontend/pom.xml @@ -0,0 +1,84 @@ + + + 4.0.0 + + + org.springframework.boot + spring-boot-starter-parent + 2.2.2.RELEASE + + + hw16_frontend + ru.otus + 2019-09-SNAPSHOT + + + + org.springframework.boot + spring-boot-starter-web + + + org.springframework.boot + spring-boot-starter-thymeleaf + + + org.springframework.boot + spring-boot-starter-websocket + + + org.springframework.integration + spring-integration-ip + + + + org.webjars + stomp-websocket + 2.3.3-1 + + + org.webjars + webjars-locator-core + + + org.webjars + sockjs-client + 1.1.2 + + + org.webjars + bootstrap + 4.3.1 + + + org.webjars + jquery + 3.1.0 + + + + hw16_common + ru.otus + 2019-09-SNAPSHOT + + + + + feService + + + org.springframework.boot + spring-boot-maven-plugin + + + org.apache.maven.plugins + maven-compiler-plugin + + 11 + 11 + + + + + diff --git a/hw16_frontend/src/main/java/ru/otus/homework/FrontApplication.java b/hw16_frontend/src/main/java/ru/otus/homework/FrontApplication.java new file mode 100644 index 0000000..4e81acc --- /dev/null +++ b/hw16_frontend/src/main/java/ru/otus/homework/FrontApplication.java @@ -0,0 +1,12 @@ +package ru.otus.homework; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; + +@SpringBootApplication +public class FrontApplication { + + public static void main(String[] args) { + SpringApplication.run(FrontApplication.class, args); + } +} diff --git a/hw16_frontend/src/main/java/ru/otus/homework/config/FrontendConfig.java b/hw16_frontend/src/main/java/ru/otus/homework/config/FrontendConfig.java new file mode 100644 index 0000000..51b2dd5 --- /dev/null +++ b/hw16_frontend/src/main/java/ru/otus/homework/config/FrontendConfig.java @@ -0,0 +1,72 @@ +package ru.otus.homework.config; + +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.event.ContextRefreshedEvent; +import org.springframework.context.event.EventListener; +import org.springframework.integration.ip.tcp.TcpInboundGateway; +import org.springframework.integration.ip.tcp.connection.TcpNetServerConnectionFactory; +import ru.otus.homework.handler.UserDataResponseHandler; +import ru.otus.homework.ms.MessageType; +import ru.otus.homework.ms.RequestHandler; +import ru.otus.homework.service.FrontendService; +import ru.otus.homework.service.FrontendServiceImpl; +import ru.otus.homework.ms.MsClient; +import ru.otus.homework.ms.MsClientImpl; +import ru.otus.homework.socket.FrontServer; +import ru.otus.homework.socket.FrontServerImp; + +@Configuration +public class FrontendConfig { + @Value("${ms.client.name}") + private String clientName; + + @Value("${ms.host}") + private String msHost; + + @Value("${ms.port}") + private Integer msPort; + + @Value("${tcp.server.port}") + private Integer tcpServerPort; + + @Bean + public MsClient frontendClient() { + return new MsClientImpl(clientName, msHost, msPort); + } + + @Bean + public FrontendService frontendService() { + return new FrontendServiceImpl(frontendClient()); + } + + @Bean + public RequestHandler responseHandler() { + return new UserDataResponseHandler(frontendService()); + } + + @Bean + public FrontServer frontServer() { + return new FrontServerImp(msg -> frontendClient().handle(msg)); + } + + @Bean + public TcpNetServerConnectionFactory server() { + return new TcpNetServerConnectionFactory(tcpServerPort); + } + + @Bean + public TcpInboundGateway inGate() { + TcpInboundGateway inGate = new TcpInboundGateway(); + inGate.setConnectionFactory(server()); + inGate.setRequestChannelName("inChannel"); + return inGate; + } + + @EventListener + public void onApplicationEvent(ContextRefreshedEvent event) { + frontendClient().addHandler(MessageType.USER_SAVE, responseHandler()); + frontendClient().addHandler(MessageType.USER_LIST, responseHandler()); + } +} diff --git a/hw16_frontend/src/main/java/ru/otus/homework/config/WebSocketConfig.java b/hw16_frontend/src/main/java/ru/otus/homework/config/WebSocketConfig.java new file mode 100644 index 0000000..015b88f --- /dev/null +++ b/hw16_frontend/src/main/java/ru/otus/homework/config/WebSocketConfig.java @@ -0,0 +1,23 @@ +package ru.otus.homework.config; + +import org.springframework.context.annotation.Configuration; +import org.springframework.messaging.simp.config.MessageBrokerRegistry; +import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker; +import org.springframework.web.socket.config.annotation.StompEndpointRegistry; +import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer; + +@Configuration +@EnableWebSocketMessageBroker +public class WebSocketConfig implements WebSocketMessageBrokerConfigurer { + + @Override + public void configureMessageBroker(MessageBrokerRegistry config) { + config.enableSimpleBroker("/topic"); + config.setApplicationDestinationPrefixes("/app"); + } + + @Override + public void registerStompEndpoints(StompEndpointRegistry registry) { + registry.addEndpoint("/ws").withSockJS(); + } +} diff --git a/hw16_frontend/src/main/java/ru/otus/homework/controller/UserController.java b/hw16_frontend/src/main/java/ru/otus/homework/controller/UserController.java new file mode 100644 index 0000000..f58cc52 --- /dev/null +++ b/hw16_frontend/src/main/java/ru/otus/homework/controller/UserController.java @@ -0,0 +1,56 @@ +package ru.otus.homework.controller; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.messaging.handler.annotation.MessageMapping; +import org.springframework.messaging.handler.annotation.Payload; +import org.springframework.messaging.simp.SimpMessagingTemplate; +import org.springframework.stereotype.Controller; +import org.springframework.web.bind.annotation.GetMapping; +import ru.otus.homework.dto.UserDto; +import ru.otus.homework.service.FrontendService; + +import java.util.List; + +@Controller +public class UserController { + private static final Logger LOG = LoggerFactory.getLogger(UserController.class); + + private final FrontendService frontendService; + private final SimpMessagingTemplate template; + + public UserController(FrontendService frontendService, SimpMessagingTemplate template) { + this.frontendService = frontendService; + this.template = template; + } + + @GetMapping("/user/save") + public String userSaveView() { + return "userSave"; + } + + @MessageMapping("/user/save") + public void saveUser(@Payload UserDto userDto) { + LOG.info("save user {}", userDto); + + frontendService.saveUser(userDto, (String message) -> { + LOG.info("response data:{}", message); + template.convertAndSend("/topic/response/user/save", message); + }); + } + + @GetMapping("/user/list") + public String userListView() { + return "userList"; + } + + @MessageMapping("/user/list") + public void userList() { + LOG.info("get list of users"); + + frontendService.getUserList((List data) -> { + LOG.info("response data:{}", data); + template.convertAndSend("/topic/response/user/list", data); + }); + } +} diff --git a/hw16_frontend/src/main/java/ru/otus/homework/handler/UserDataResponseHandler.java b/hw16_frontend/src/main/java/ru/otus/homework/handler/UserDataResponseHandler.java new file mode 100644 index 0000000..ab7bc79 --- /dev/null +++ b/hw16_frontend/src/main/java/ru/otus/homework/handler/UserDataResponseHandler.java @@ -0,0 +1,50 @@ +package ru.otus.homework.handler; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import ru.otus.homework.common.Serializers; +import ru.otus.homework.dto.UserDto; +import ru.otus.homework.exception.DataHandlerException; +import ru.otus.homework.ms.Message; +import ru.otus.homework.ms.RequestHandler; +import ru.otus.homework.service.FrontendService; + +import java.util.List; +import java.util.Optional; +import java.util.UUID; + +public class UserDataResponseHandler implements RequestHandler { + private static final Logger LOG = LoggerFactory.getLogger(UserDataResponseHandler.class); + private final FrontendService frontendService; + + public UserDataResponseHandler(FrontendService frontendService) { + this.frontendService = frontendService; + } + + @Override + public Optional handle(Message msg) { + LOG.info("new message:{}", msg); + try { + UUID sourceMessageId = msg.getSourceMessageId(). + orElseThrow(() -> new RuntimeException("Not found sourceMsg for message:" + msg.getId())); + String messageType = msg.getType(); + switch (messageType) { + case "UserSave": + String message = Serializers.deserialize(msg.getPayload(), String.class); + frontendService.takeConsumer(sourceMessageId, String.class). + ifPresent(consumer -> consumer.accept(message)); + break; + case "UserList": + List users = Serializers.deserialize(msg.getPayload(), List.class); + frontendService.takeConsumer(sourceMessageId, List.class). + ifPresent(consumer -> consumer.accept(users)); + break; + default: + throw new DataHandlerException(String.format("Unknown message type %s", messageType)); + } + } catch (Exception ex) { + LOG.error("msg: {}", ex.getMessage()); + } + return Optional.empty(); + } +} diff --git a/hw16_frontend/src/main/java/ru/otus/homework/service/FrontendService.java b/hw16_frontend/src/main/java/ru/otus/homework/service/FrontendService.java new file mode 100644 index 0000000..3489a09 --- /dev/null +++ b/hw16_frontend/src/main/java/ru/otus/homework/service/FrontendService.java @@ -0,0 +1,17 @@ +package ru.otus.homework.service; + +import ru.otus.homework.dto.UserDto; + +import java.util.List; +import java.util.Optional; +import java.util.UUID; +import java.util.function.Consumer; + +public interface FrontendService { + + void saveUser(UserDto userDto, Consumer dataConsumer); + + void getUserList(Consumer> dataConsumer); + + Optional> takeConsumer(UUID sourceMessageId, Class tClass); +} diff --git a/hw16_frontend/src/main/java/ru/otus/homework/service/FrontendServiceImpl.java b/hw16_frontend/src/main/java/ru/otus/homework/service/FrontendServiceImpl.java new file mode 100644 index 0000000..701ccb9 --- /dev/null +++ b/hw16_frontend/src/main/java/ru/otus/homework/service/FrontendServiceImpl.java @@ -0,0 +1,50 @@ +package ru.otus.homework.service; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import ru.otus.homework.dto.UserDto; +import ru.otus.homework.ms.Message; +import ru.otus.homework.ms.MessageType; +import ru.otus.homework.ms.MsClient; + +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Consumer; + +public class FrontendServiceImpl implements FrontendService { + private static final Logger LOG = LoggerFactory.getLogger(FrontendServiceImpl.class); + private final Map> consumerMap = new ConcurrentHashMap<>(); + + private final MsClient msClient; + + public FrontendServiceImpl(MsClient msClient) { + this.msClient = msClient; + } + + @Override + public void saveUser(UserDto userDto, Consumer dataConsumer) { + Message outMsg = msClient.produceMessage(null, userDto, MessageType.USER_SAVE); + consumerMap.put(outMsg.getId(), dataConsumer); + msClient.sendMessage(outMsg); + } + + @Override + public void getUserList(Consumer> dataConsumer) { + Message outMsg = msClient.produceMessage(null, null, MessageType.USER_LIST); + consumerMap.put(outMsg.getId(), dataConsumer); + msClient.sendMessage(outMsg); + } + + @Override + public Optional> takeConsumer(UUID sourceMessageId, Class tClass) { + Consumer consumer = (Consumer) consumerMap.remove(sourceMessageId); + if (consumer == null) { + LOG.warn("consumer not found for:{}", sourceMessageId); + return Optional.empty(); + } + return Optional.of(consumer); + } +} diff --git a/hw16_frontend/src/main/java/ru/otus/homework/socket/FrontServer.java b/hw16_frontend/src/main/java/ru/otus/homework/socket/FrontServer.java new file mode 100644 index 0000000..8fbb57a --- /dev/null +++ b/hw16_frontend/src/main/java/ru/otus/homework/socket/FrontServer.java @@ -0,0 +1,5 @@ +package ru.otus.homework.socket; + +public interface FrontServer { + void listen(byte[] in); +} diff --git a/hw16_frontend/src/main/java/ru/otus/homework/socket/FrontServerImp.java b/hw16_frontend/src/main/java/ru/otus/homework/socket/FrontServerImp.java new file mode 100644 index 0000000..a47afd3 --- /dev/null +++ b/hw16_frontend/src/main/java/ru/otus/homework/socket/FrontServerImp.java @@ -0,0 +1,33 @@ +package ru.otus.homework.socket; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.integration.annotation.ServiceActivator; +import ru.otus.homework.ms.Message; + +import java.io.ByteArrayInputStream; +import java.io.ObjectInputStream; +import java.util.function.Consumer; + +public class FrontServerImp implements FrontServer { + private static final Logger LOG = LoggerFactory.getLogger(FrontServerImp.class); + + private final Consumer consumer; + + public FrontServerImp(Consumer consumer) { + this.consumer = consumer; + } + + @ServiceActivator(inputChannel = "inChannel") + public void listen(byte[] in) { + LOG.info("front server received {} bytes", in.length); + try { + ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(in)); + Message msg = (Message) ois.readObject(); + LOG.info("handle message {}", msg.getId()); + consumer.accept(msg); + } catch (Exception e) { + LOG.error("error for front server", e); + } + } +} diff --git a/hw16_frontend/src/main/resources/application-fe2.yml b/hw16_frontend/src/main/resources/application-fe2.yml new file mode 100644 index 0000000..597b18d --- /dev/null +++ b/hw16_frontend/src/main/resources/application-fe2.yml @@ -0,0 +1,8 @@ +server: + port: 8091 + +tcp.server.port: 9010 + +ms: + client: + name: frontend2 diff --git a/hw16_frontend/src/main/resources/application.yml b/hw16_frontend/src/main/resources/application.yml new file mode 100644 index 0000000..27bbd06 --- /dev/null +++ b/hw16_frontend/src/main/resources/application.yml @@ -0,0 +1,21 @@ +server: + port: 8090 + +tcp.server.port: 9009 +tcp.server.autostart: true + +ms: + host: localhost + port: 8081 + client: + name: frontend1 + +spring: + mvc: + throw-exception-if-no-handler-found: true + +logging.level: + ROOT: INFO + org.springframework: INFO + org.springframework.web: INFO + org.apache: INFO diff --git a/hw16_frontend/src/main/resources/logback.xml b/hw16_frontend/src/main/resources/logback.xml new file mode 100644 index 0000000..31883bd --- /dev/null +++ b/hw16_frontend/src/main/resources/logback.xml @@ -0,0 +1,13 @@ + + + + + %d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n + + + + + + + + diff --git a/hw16_frontend/src/main/resources/static/list.js b/hw16_frontend/src/main/resources/static/list.js new file mode 100644 index 0000000..8756091 --- /dev/null +++ b/hw16_frontend/src/main/resources/static/list.js @@ -0,0 +1,27 @@ +var stompClient = null; + +function getUserList() { + stompClient = Stomp.over(new SockJS('/ws')); + + stompClient.connect({}, function () { + console.log('connected stompClient'); + stompClient.send("/app/user/list", {}, ''); + stompClient.subscribe('/topic/response/user/list', function (output) { + console.log('response from /user/list'); + var usersResp = JSON.parse(output.body); + + var tableData = '' + for (i in usersResp) { + tableData+= ''; + tableData+= '' + usersResp[i].name + ''; + tableData+= '' + usersResp[i].login + ''; + tableData+= '' + usersResp[i].role + ''; + tableData+= ''; + } + tableData+=''; + document.getElementById('tableData').innerHTML = tableData; + }); + }, function (err) { + alert('error' + err); + }); +} diff --git a/hw16_frontend/src/main/resources/static/save.js b/hw16_frontend/src/main/resources/static/save.js new file mode 100644 index 0000000..c5c7535 --- /dev/null +++ b/hw16_frontend/src/main/resources/static/save.js @@ -0,0 +1,31 @@ +var stompClient = null; + +function connect() { + stompClient = Stomp.over(new SockJS('/ws')); + stompClient.connect({}, function () { + console.log('connected stompClient'); + stompClient.subscribe('/topic/response/user/save', function (output) { + console.log('response from user/save'); + alert(output.body); + }); + }, function (err) { + alert('error' + err); + }); +} + +function saveUser() { + var name = $("#name").val(); + var login = $("#login").val(); + var password = $("#password").val(); + var role = $("#role").val(); + + if(!login || 0 === login.length || !password || 0 === password.length){ + alert('Login and password can not be null'); + }else{ + stompClient.send("/app/user/save", {}, JSON.stringify({name, login, password, role})); + } +}; + +$(() => { + $("form").on('submit', event => event.preventDefault()); +}); diff --git a/hw16_frontend/src/main/resources/templates/error.html b/hw16_frontend/src/main/resources/templates/error.html new file mode 100644 index 0000000..3a2d75e --- /dev/null +++ b/hw16_frontend/src/main/resources/templates/error.html @@ -0,0 +1,13 @@ + + + + Error + + +
+

Error

+

+ List of all users +
+ + diff --git a/hw16_frontend/src/main/resources/templates/userList.html b/hw16_frontend/src/main/resources/templates/userList.html new file mode 100644 index 0000000..1040b19 --- /dev/null +++ b/hw16_frontend/src/main/resources/templates/userList.html @@ -0,0 +1,38 @@ + + + + + + Users + + + + + + + + + + + +
+

Users

+ + + + + + + + + +
NameLoginRole
+
+

+
+
+ + diff --git a/hw16_frontend/src/main/resources/templates/userSave.html b/hw16_frontend/src/main/resources/templates/userSave.html new file mode 100644 index 0000000..9f024f4 --- /dev/null +++ b/hw16_frontend/src/main/resources/templates/userSave.html @@ -0,0 +1,61 @@ + + + + + + Admin page + + + + + + + + + + + +
+

Save new user:

+
+
+
+
+
+ + +
+
+ + +
+
+ + +
+
+ +
+ +
+
+
+
+ +
+
+
+

+ List of all users +
+
+
+ + diff --git a/hw16_messageServer/README.md b/hw16_messageServer/README.md new file mode 100644 index 0000000..963475b --- /dev/null +++ b/hw16_messageServer/README.md @@ -0,0 +1,34 @@ +# MessageServer + Cервер из предыдущего ДЗ про MessageSystem разделить на три приложения: + - MessageServer + - Frontend + - DBServer + + Сделать MessageServer сокет-сервером, Frontend и DBServer клиентами. + Пересылать сообщения с Frontend на DBService через MessageServer. + Запустить приложение с двумя Frontend и двумя DBService (но на одной базе + данных) на разных портах. + Frontend и DBService запускать "руками". + По желанию, можно сделать запуск Frontend и DBServer из MessageServer. + Такой запуск должен быть "отчуждаемый", т.е. "сборка" должна запускаться на + другом компьютере без особых хлопот. + + +**Конфигурация портов, которая используется в решении:** + ``` + websocket(8090) --> frontend1(9009) database1(8083) + --> MessageServer(8081)--> + websocket(8091) --> frontend2(9010) database2(8084) +``` +**Запуск сервисов из корня проекта** +- java -jar hw16_frontend/target/feService.jar (**frontend1**) +- java -jar hw16_frontend/target/feService.jar --spring.profiles.active=fe2 (**frontend2**) + +- java -jar hw16_dbServer/target/dbService.jar (**database1**) +- java -jar hw16_dbServer/target/dbService.jar --spring.profiles.active=db2 (**database2**) + +- java -jar hw16_messageServer/target/msServer.jar (**message server**) + +**Сохранение пользователей** +- http://localhost:8090/user/save +- http://localhost:8091/user/save \ No newline at end of file diff --git a/hw16_messageServer/pom.xml b/hw16_messageServer/pom.xml new file mode 100644 index 0000000..39cab1a --- /dev/null +++ b/hw16_messageServer/pom.xml @@ -0,0 +1,47 @@ + + + 4.0.0 + + + org.springframework.boot + spring-boot-starter-parent + 2.2.2.RELEASE + + + hw16_messageServer + ru.otus + 2019-09-SNAPSHOT + + + + org.springframework.boot + spring-boot-starter + + + + hw16_common + ru.otus + 2019-09-SNAPSHOT + + + + + msServer + + + org.springframework.boot + spring-boot-maven-plugin + + + org.apache.maven.plugins + maven-compiler-plugin + + 11 + 11 + + + + + diff --git a/hw16_messageServer/src/main/java/ru/otus/homework/MsApplication.java b/hw16_messageServer/src/main/java/ru/otus/homework/MsApplication.java new file mode 100644 index 0000000..c244350 --- /dev/null +++ b/hw16_messageServer/src/main/java/ru/otus/homework/MsApplication.java @@ -0,0 +1,16 @@ +package ru.otus.homework; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import ru.otus.homework.config.DbClientsProperties; +import ru.otus.homework.config.FeClientsProperties; + +@SpringBootApplication +@EnableConfigurationProperties({DbClientsProperties.class, FeClientsProperties.class}) +public class MsApplication { + + public static void main(String[] args) { + SpringApplication.run(MsApplication.class, args); + } +} diff --git a/hw16_messageServer/src/main/java/ru/otus/homework/config/DbClientsProperties.java b/hw16_messageServer/src/main/java/ru/otus/homework/config/DbClientsProperties.java new file mode 100644 index 0000000..e889fb3 --- /dev/null +++ b/hw16_messageServer/src/main/java/ru/otus/homework/config/DbClientsProperties.java @@ -0,0 +1,20 @@ +package ru.otus.homework.config; + +import org.springframework.boot.context.properties.ConfigurationProperties; + +import java.util.HashMap; +import java.util.Map; + +@ConfigurationProperties(prefix = "dbclients") +public class DbClientsProperties { + + /** + * key is name of the client, value - its host anp port separated by ":" + * Example: "database1: localhost:8083" + */ + private final Map map = new HashMap<>(); + + public Map getMap() { + return map; + } +} diff --git a/hw16_messageServer/src/main/java/ru/otus/homework/config/FeClientsProperties.java b/hw16_messageServer/src/main/java/ru/otus/homework/config/FeClientsProperties.java new file mode 100644 index 0000000..bfa03d7 --- /dev/null +++ b/hw16_messageServer/src/main/java/ru/otus/homework/config/FeClientsProperties.java @@ -0,0 +1,20 @@ +package ru.otus.homework.config; + +import org.springframework.boot.context.properties.ConfigurationProperties; + +import java.util.HashMap; +import java.util.Map; + +@ConfigurationProperties(prefix = "feclients") +public class FeClientsProperties { + + /** + * key is name of the client, value - its host anp port separated by ":" + * Example: "frontend1: localhost:9009" + */ + private final Map map = new HashMap<>(); + + public Map getMap() { + return map; + } +} diff --git a/hw16_messageServer/src/main/java/ru/otus/homework/config/MessageSystemConfig.java b/hw16_messageServer/src/main/java/ru/otus/homework/config/MessageSystemConfig.java new file mode 100644 index 0000000..0eb6a48 --- /dev/null +++ b/hw16_messageServer/src/main/java/ru/otus/homework/config/MessageSystemConfig.java @@ -0,0 +1,59 @@ +package ru.otus.homework.config; + +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.event.ContextRefreshedEvent; +import org.springframework.context.event.EventListener; +import ru.otus.homework.exception.DataHandlerException; +import ru.otus.homework.messagesystem.MessageSystem; +import ru.otus.homework.messagesystem.MessageSystemImpl; +import ru.otus.homework.ms.MsClientImpl; +import ru.otus.homework.socket.ListenServer; +import ru.otus.homework.socket.ListenServerImpl; + +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +@Configuration +public class MessageSystemConfig { + @Value("${server.port}") + private Integer serverPort; + + private final DbClientsProperties dbClientsProperties; + private final FeClientsProperties feClientsProperties; + + public MessageSystemConfig(DbClientsProperties dbClientsProperties, FeClientsProperties feClientsProperties) { + this.dbClientsProperties = dbClientsProperties; + this.feClientsProperties = feClientsProperties; + } + + @Bean + public MessageSystem messageSystem() { + return new MessageSystemImpl(); + } + + @Bean + public ListenServer messageServer() { + return new ListenServerImpl(serverPort, msg -> messageSystem().newMessage(msg)); + } + + @EventListener + public void onApplicationEvent(ContextRefreshedEvent event) { + convertMap(dbClientsProperties.getMap()).forEach(msClient -> messageSystem().addDbClient(msClient)); + convertMap(feClientsProperties.getMap()).forEach(msClient -> messageSystem().addFeClient(msClient)); + messageServer().start(); + } + + private List convertMap(Map map) { + return map.entrySet().stream().map(entry -> { + try { + String[] address = entry.getValue().split(":"); + return new MsClientImpl(entry.getKey(), address[0], Integer.parseInt(address[1])); + } catch (Exception e) { + throw new DataHandlerException("clients initialization failed", e); + } + }).collect(Collectors.toList()); + } +} diff --git a/hw16_messageServer/src/main/java/ru/otus/homework/messagesystem/MessageSystem.java b/hw16_messageServer/src/main/java/ru/otus/homework/messagesystem/MessageSystem.java new file mode 100644 index 0000000..983be30 --- /dev/null +++ b/hw16_messageServer/src/main/java/ru/otus/homework/messagesystem/MessageSystem.java @@ -0,0 +1,24 @@ +package ru.otus.homework.messagesystem; + +import ru.otus.homework.ms.Message; +import ru.otus.homework.ms.MsClient; + +public interface MessageSystem { + + void addDbClient(MsClient msClient); + + void addFeClient(MsClient msClient); + + void removeClient(String clientId); + + boolean newMessage(Message msg); + + void dispose() throws InterruptedException; + + void dispose(Runnable callback) throws InterruptedException; + + void start(); + + int currentQueueSize(); +} + diff --git a/hw16_messageServer/src/main/java/ru/otus/homework/messagesystem/MessageSystemImpl.java b/hw16_messageServer/src/main/java/ru/otus/homework/messagesystem/MessageSystemImpl.java new file mode 100644 index 0000000..4dceca5 --- /dev/null +++ b/hw16_messageServer/src/main/java/ru/otus/homework/messagesystem/MessageSystemImpl.java @@ -0,0 +1,184 @@ +package ru.otus.homework.messagesystem; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import ru.otus.homework.ms.Message; +import ru.otus.homework.ms.MsClient; + +import java.util.ArrayDeque; +import java.util.Deque; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +public final class MessageSystemImpl implements MessageSystem { + private static final Logger LOG = LoggerFactory.getLogger(MessageSystemImpl.class); + private static final int MESSAGE_QUEUE_SIZE = 100_000; + private static final int MSG_HANDLER_THREAD_LIMIT = 100; + + private final AtomicBoolean runFlag = new AtomicBoolean(true); + + private final Deque dbClientList = new ArrayDeque<>(); + private final Map feClientMap = new ConcurrentHashMap<>(); + private final BlockingQueue messageQueue = new ArrayBlockingQueue<>(MESSAGE_QUEUE_SIZE); + + private Runnable disposeCallback; + + private final ExecutorService msgProcessor = Executors.newSingleThreadExecutor(runnable -> { + Thread thread = new Thread(runnable); + thread.setName("msg-processor-thread"); + return thread; + }); + + private final ExecutorService msgHandler = + Executors.newFixedThreadPool(MSG_HANDLER_THREAD_LIMIT, new ThreadFactory() { + private final AtomicInteger threadNameSeq = new AtomicInteger(0); + + @Override + public Thread newThread(Runnable runnable) { + Thread thread = new Thread(runnable); + thread.setName("msg-handler-thread-" + threadNameSeq.incrementAndGet()); + return thread; + } + }); + + public MessageSystemImpl() { + start(); + } + + public MessageSystemImpl(boolean startProcessing) { + if (startProcessing) { + start(); + } + } + + @Override + public void start() { + msgProcessor.submit(this::msgProcessor); + } + + @Override + public int currentQueueSize() { + return messageQueue.size(); + } + + @Override + public void addDbClient(MsClient msClient) { + LOG.info("new database client: {}", msClient.getName()); + dbClientList.add(msClient); + } + + @Override + public void addFeClient(MsClient msClient) { + LOG.info("new frontend client: {}", msClient.getName()); + if (feClientMap.containsKey(msClient.getName())) { + throw new IllegalArgumentException("Error, client: " + msClient.getName() + " already exists"); + } + feClientMap.put(msClient.getName(), msClient); + } + + @Override + public void removeClient(String clientId) { + Optional dbClient = dbClientList.stream(). + filter(msClient -> clientId.equals(msClient.getName())).findAny(); + dbClient.ifPresent(dbClientList::remove); + if (feClientMap.remove(clientId) != null || dbClient.isPresent()) { + LOG.info("client {} removed", clientId); + } else { + LOG.warn("client not found: {}", clientId); + } + } + + @Override + public boolean newMessage(Message msg) { + LOG.info("new message: {}", msg.getId()); + if (runFlag.get()) { + return messageQueue.offer(msg); + } else { + LOG.warn("MS is being shutting down... rejected:{}", msg); + return false; + } + } + + @Override + public void dispose() throws InterruptedException { + LOG.info("now in the messageQueue {} messages", currentQueueSize()); + runFlag.set(false); + insertStopMessage(); + msgProcessor.shutdown(); + msgHandler.awaitTermination(60, TimeUnit.SECONDS); + } + + @Override + public void dispose(Runnable callback) throws InterruptedException { + disposeCallback = callback; + dispose(); + } + + private void insertStopMessage() throws InterruptedException { + boolean result = messageQueue.offer(Message.VOID_MESSAGE); + while (!result) { + Thread.sleep(100); + result = messageQueue.offer(Message.VOID_MESSAGE); + } + } + + private void messageHandlerShutdown() { + msgHandler.shutdown(); + LOG.info("msgHandler has been shut down"); + } + + private void msgProcessor() { + LOG.info("msgProcessor started, {}", currentQueueSize()); + while (runFlag.get() || !messageQueue.isEmpty()) { + try { + Message msg = messageQueue.take(); + LOG.info("MSG_PROCESSOR"); + if (msg == Message.VOID_MESSAGE) { + LOG.info("received the stop message"); + } else { + handleMessage(msg); + } + } catch (InterruptedException ex) { + LOG.error(ex.getMessage(), ex); + Thread.currentThread().interrupt(); + } catch (Exception ex) { + LOG.error(ex.getMessage(), ex); + } + } + + if (disposeCallback != null) { + msgHandler.submit(disposeCallback); + } + msgHandler.submit(this::messageHandlerShutdown); + LOG.info("msgProcessor finished"); + } + + private void handleMessage(Message msg) { + MsClient clientTo = gerClientTo(msg.getTo()); + if (clientTo != null) { + msgHandler.submit(() -> clientTo.sendMessage(msg)); + } else { + LOG.warn("client not found"); + } + } + + private MsClient gerClientTo(String msgClientTo) { + MsClient clientTo; + if (msgClientTo == null && !dbClientList.isEmpty()) { + clientTo = dbClientList.removeFirst(); + dbClientList.addLast(clientTo); + } else { + clientTo = feClientMap.get(msgClientTo); + } + return clientTo; + } +} diff --git a/hw16_messageServer/src/main/resources/application.yml b/hw16_messageServer/src/main/resources/application.yml new file mode 100644 index 0000000..47ba4b3 --- /dev/null +++ b/hw16_messageServer/src/main/resources/application.yml @@ -0,0 +1,16 @@ +server: + port: 8081 + +dbclients: + map: + database1: localhost:8083 + database2: localhost:8084 + +feclients: + map: + frontend1: localhost:9009 + frontend2: localhost:9010 + +logging.level: + ROOT: INFO + org.springframework: INFO diff --git a/hw16_messageServer/src/main/resources/logback.xml b/hw16_messageServer/src/main/resources/logback.xml new file mode 100644 index 0000000..31883bd --- /dev/null +++ b/hw16_messageServer/src/main/resources/logback.xml @@ -0,0 +1,13 @@ + + + + + %d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n + + + + + + + + diff --git a/pom.xml b/pom.xml index 0f8adcf..5242e0f 100644 --- a/pom.xml +++ b/pom.xml @@ -16,6 +16,7 @@ hw01-maven hw02-arrayList + hw04-logging hw05-testFramework hw06-atm hw07-atm-department @@ -26,6 +27,11 @@ hw12-webServer hw13-DI hw14-threads + hw15-messageSystem + hw16_messageServer + hw16_frontend + hw16_dbServer + hw16_common