tests reactor

zacscoding · zacscoding · commit 92e9874ee2ef · 2019-08-10T00:19:34.000+09:00
diff --git a/springboot-context-demo/scripts/prototype.sh b/springboot-context-demo/scripts/prototype.sh
@@ -12,8 +12,10 @@ function printHelp() {
   echo "commands :"
   echo "  get       getting bean all or one"
   echo "    e.g : ${SCRIPT_NAME} get 1 or ${SCRIPT_NAME} get"
-  echo "  getctx    getting bean all from context"
+  echo "  getctx    getting prototype bean all from context"
   echo "    e.g : ${SCRIPT_NAME} getctx"
+  echo "  getbeans  getting beans all from context"
+  echo "    e.g : ${SCRIPT_NAME} getbeans"
   echo "  create    create prototype bean"
   echo "    e.g : ${SCRIPT_NAME} create "
   echo "  delete    destroy prototype bean"
@@ -35,6 +37,11 @@ function getctx() {
   echo ""
 }
 
+function getBeans() {
+  curl -X GET -H "Content-Type: application/json; charset=utf-8" ${URL}/prototypes/beans
+  echo ""
+}
+
 function create() {
   curl -X POST -H "Content-Type: application/json; charset=utf-8" ${URL}/prototype
   echo ""
@@ -47,7 +54,7 @@ function delete() {
 
 # command
 case "${1}" in
-  get | getctx | create | delete )
+  get | getctx | getbeans | create | delete )
     COMMAND=${1}
     ;;
   * )
@@ -64,6 +71,9 @@ case "${COMMAND}" in
   getctx )
     getctx
   ;;
+  getbeans )
+    getBeans
+  ;;
   create )
     create
   ;;
diff --git a/springboot-context-demo/src/main/java/demo/proto/PrototypeBeanController.java b/springboot-context-demo/src/main/java/demo/proto/PrototypeBeanController.java
@@ -45,7 +45,7 @@ public ResponseEntity getBeans() {
     }
 
     @GetMapping("/prototypes/ctx")
-    public ResponseEntity getBeansFromCtx() {
+    public ResponseEntity getPrototypeBeansFromCtx() {
         // 매번 새로운 PrototypeBean1 이 생성되어 Map::size() == 1 인 맵을 반환
         StringBuilder result = new StringBuilder("Getting PrototypeBeans1 from ctx\n");
 
@@ -60,6 +60,25 @@ public ResponseEntity getBeansFromCtx() {
         return ResponseEntity.ok(result);
     }
 
+
+    @GetMapping("/prototypes/beans")
+    public ResponseEntity getBeansFromCtx() {
+        StringBuilder result = new StringBuilder();
+
+        String[] beanNames = ctx.getBeanDefinitionNames();
+        for (String beanName : beanNames) {
+            Object bean = ctx.getBean(beanName);
+            if (!bean.getClass().getName().startsWith("demo")) {
+                continue;
+            }
+            String display = String.format("%-30s : %-40s [%s]", beanName, bean.getClass().getName(), bean.toString());
+            result.append(display)
+                .append("\n");
+        }
+
+        return ResponseEntity.ok(result.toString());
+    }
+
     @GetMapping("/prototype/{id}")
     public ResponseEntity getBean(@PathVariable("id") Long id) {
         logger.info("Try to get bean id : {}", id);
@@ -98,20 +117,11 @@ public ResponseEntity destroyBean(@PathVariable("id") Long id) {
             return ResponseEntity.notFound().build();
         }
 
-        String result = null;
+        bean = null;
 
-        try {
-            result = bean.toString();
+        System.gc();
+        System.runFinalization();
 
-            if (bean instanceof DisposableBean) {
-                logger.info("Can cast DisposableBean");
-                DisposableBean disposable = (DisposableBean) bean;
-                disposable.destroy();
-            }
-        } catch (Exception e) {
-            result = e.getMessage();
-        }
-
-        return ResponseEntity.ok(result);
+        return ResponseEntity.ok().build();
     }
 }
diff --git a/springboot-reactor-demo/pom.xml b/springboot-reactor-demo/pom.xml
@@ -24,20 +24,21 @@
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-aop</artifactId>
         </dependency>
-        <dependency>
+        <!--<dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-web</artifactId>
-        </dependency>
-        <!--
+        </dependency>-->
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-webflux</artifactId>
-        </dependency>-->
-        <dependency>
-            <groupId>io.projectreactor</groupId>
-            <artifactId>reactor-bus</artifactId>
         </dependency>
 
+        <!-- https://mvnrepository.com/artifact/io.projectreactor/reactor-core -->
+        <!--<dependency>
+            <groupId>io.projectreactor</groupId>
+            <artifactId>reactor-core</artifactId>
+            <version>3.2.11.RELEASE</version>
+        </dependency>-->
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-devtools</artifactId>
diff --git a/springboot-reactor-demo/src/main/java/demo/DemoApplication.java b/springboot-reactor-demo/src/main/java/demo/DemoApplication.java
@@ -1,13 +1,9 @@
 package demo;
 
-import javax.servlet.http.HttpServletRequest;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
-import org.springframework.web.bind.annotation.GetMapping;
-import org.springframework.web.bind.annotation.PathVariable;
 import org.springframework.web.bind.annotation.RestController;
-import org.springframework.web.servlet.HandlerMapping;
 
 @Slf4j
 @RestController
@@ -18,10 +14,10 @@ public static void main(String[] args) {
         SpringApplication.run(DemoApplication.class, args);
     }
 
-    @GetMapping("/main/{pathVariable}")
+    /*@GetMapping("/main/{pathVariable}")
     public void temp(@PathVariable String pathVariable, HttpServletRequest request) {
         String attr = (String) request.getAttribute(HandlerMapping.BEST_MATCHING_PATTERN_ATTRIBUTE);
         logger.info("Request /main page. pathVariable : {}, attr : {} / uri : {} / url : {}"
             , pathVariable, attr, request.getRequestURI(), request.getRequestURL());
-    }
+    }*/
 }
diff --git a/springboot-reactor-demo/src/main/java/demo/dto/Pair.java b/springboot-reactor-demo/src/main/java/demo/dto/Pair.java
@@ -0,0 +1,24 @@
+package demo.dto;
+
+import lombok.Getter;
+import lombok.ToString;
+
+/**
+ *
+ */
+@Getter
+@ToString
+public class Pair<F, S> {
+
+    private F first;
+    private S second;
+
+    public static <F, S> Pair<F, S> newInstance(F first, S second) {
+        return new Pair<>(first, second);
+    }
+
+    private Pair(F first, S second) {
+        this.first = first;
+        this.second = second;
+    }
+}
diff --git a/springboot-reactor-demo/src/main/java/demo/publisher1/BlockProcessor.java b/springboot-reactor-demo/src/main/java/demo/publisher1/BlockProcessor.java
@@ -0,0 +1,28 @@
+package demo.publisher1;
+
+import demo.dto.Pair;
+import lombok.extern.slf4j.Slf4j;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.FluxSink;
+import reactor.core.publisher.ReplayProcessor;
+
+@Slf4j(topic = "publisher1.BlockProcessor")
+public class BlockProcessor {
+
+    // replay only last item
+    private final ReplayProcessor<Pair<Long, String>> blockProcessor = ReplayProcessor.cacheLast();
+    private final FluxSink<Pair<Long, String>> blockSink = blockProcessor.sink();
+
+    public Publisher<Pair<Long, String>> blockStream() {
+        return blockProcessor;
+    }
+
+    public void newBlock(Pair<Long, String> block) {
+        logger.info("receive new block : {}", block);
+        blockSink.next(block);
+    }
+
+    public void error(Throwable throwable) {
+        blockSink.error(throwable);
+    }
+}
diff --git a/springboot-reactor-demo/src/main/java/demo/publisher1/MockBlockchain.java b/springboot-reactor-demo/src/main/java/demo/publisher1/MockBlockchain.java
@@ -0,0 +1,39 @@
+package demo.publisher1;
+
+import demo.dto.Pair;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ *
+ */
+public class MockBlockchain implements Runnable {
+
+    private AtomicLong blockNumber = new AtomicLong(0L);
+    private BlockProcessor blockProcessor;
+
+    public MockBlockchain(BlockProcessor blockProcessor) {
+        this.blockProcessor = blockProcessor;
+    }
+
+    @Override
+    public void run() {
+        try {
+            while (!Thread.currentThread().isInterrupted()) {
+                Long bestBlockNumber = blockNumber.getAndIncrement();
+
+                if (bestBlockNumber > 5) {
+                    blockProcessor.error(new Exception("Force exception"));
+                } else {
+                    Pair<Long, String> newBlock = Pair.newInstance(bestBlockNumber, UUID.randomUUID().toString());
+                    blockProcessor.newBlock(newBlock);
+                }
+
+                TimeUnit.SECONDS.sleep(2L);
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+}
diff --git a/springboot-reactor-demo/src/main/java/demo/util/ThreadUtil.java b/springboot-reactor-demo/src/main/java/demo/util/ThreadUtil.java
@@ -0,0 +1,62 @@
+package demo.util;
+
+import java.io.PrintStream;
+
+/**
+ *
+ */
+public class ThreadUtil {
+
+    private static String NEW_LINE = System.getProperty("line.separator");
+    private static PrintStream PS = System.out;
+
+    public static String getStackTraceString(int cursor) {
+        StackTraceElement[] elts = Thread.currentThread().getStackTrace();
+        if (elts == null || elts.length == 1) {
+            return "";
+        }
+
+        StringBuilder sb = new StringBuilder();
+        int start, size;
+        if (cursor >= 0) {
+            start = cursor + 2;
+            size = elts.length;
+        } else {
+            start = 2;
+            size = start - cursor + 1;
+        }
+
+        return getStackTraceString(elts, start, size);
+    }
+
+    public static String getStackTraceString(StackTraceElement[] se, int start, int size) {
+        if (se == null) {
+            return "";
+        }
+
+        if (size < 0) {
+            size = 0;
+        }
+        size = Math.min(size, se.length);
+        if (start >= size) {
+            return "";
+        }
+
+        StringBuilder sb = new StringBuilder();
+        for (int i = start; i < size; i++) {
+            sb.append("\t").append(se[i].toString());
+            if (i != size - 1) {
+                sb.append(NEW_LINE);
+            }
+        }
+
+        return sb.toString();
+    }
+
+    public static void printStackTrace() {
+        PS.println(getStackTraceString(2));
+    }
+
+    private ThreadUtil() {
+    }
+}
diff --git a/springboot-reactor-demo/src/test/java/demo/publisher1/Publisher1Tests.java b/springboot-reactor-demo/src/test/java/demo/publisher1/Publisher1Tests.java
@@ -0,0 +1,53 @@
+package demo.publisher1;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import demo.dto.Pair;
+import demo.util.ThreadUtil;
+import java.util.concurrent.CountDownLatch;
+import javax.swing.TransferHandler;
+import org.junit.Test;
+import reactor.core.Disposable;
+import reactor.core.publisher.Flux;
+
+/**
+ *
+ */
+public class Publisher1Tests {
+
+    CountDownLatch countDownLatch = new CountDownLatch(1);
+    Long lastBlock;
+
+    @Test
+    public void runTests() throws Exception {
+        BlockProcessor blockProcessor = new BlockProcessor();
+        Disposable disposable = Flux.from(blockProcessor.blockStream())
+            .subscribe(this::insertBlock, this::onError);
+
+        Thread t = new Thread(new MockBlockchain(blockProcessor));
+        t.setDaemon(true);
+        t.start();
+
+        countDownLatch.await();
+        System.out.println("## after complete.. :: " + disposable.isDisposed());
+
+        Flux.from(blockProcessor.blockStream())
+            .subscribe(pair -> {
+                    assertThat(pair.getFirst()).isEqualTo(lastBlock);
+                }
+            );
+    }
+
+    private void insertBlock(Pair<Long, String> newBlock) {
+        System.out.println("## [Subscriber] Receive new block : " + newBlock);
+        lastBlock = newBlock.getFirst();
+        //ThreadUtil.printStackTrace();
+        // countDownLatch.countDown();
+    }
+
+    private void onError(Throwable throwable) {
+        System.err.println("exception occur : " + throwable.getMessage());
+        countDownLatch.countDown();
+    }
+
+}
diff --git a/springboot-reactor-demo/src/test/java/demo/reactor/BasicLearnTest.java b/springboot-reactor-demo/src/test/java/demo/reactor/BasicLearnTest.java
@@ -7,19 +7,24 @@
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.function.Consumer;
 import org.junit.Test;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.GroupedFlux;
 import reactor.core.publisher.Mono;
 
 /**
- * @GitHub : https://github.com/zacscoding
+ *
  */
 public class BasicLearnTest {
 
     @Test
     public void flux() throws Exception {
         Flux<String> just = Flux.just("1", "2", "3");
+        Consumer<String> consumer1 = (value) -> System.out.println("[Consumer1] " + value);
+        Consumer<String> consumer2 = (value) -> System.out.println("[Consumer2] " + value);
+        just.subscribe(consumer1);
+        just.subscribe(consumer2);
     }
 
     @Test
diff --git a/springboot-reactor-demo/src/test/java/demo/reactor/FluxTest.java b/springboot-reactor-demo/src/test/java/demo/reactor/FluxTest.java
