log processing errors

bsideup · May 14, 2018 · 55830a9 · 55830a9
1 parent 106b302
commit 55830a9
Showing 1 changed file with 12 additions and 5 deletions.
diff --git a/app/src/main/java/com/github/bsideup/liiklus/service/ReactorLiiklusServiceImpl.java b/app/src/main/java/com/github/bsideup/liiklus/service/ReactorLiiklusServiceImpl.java
@@ -59,7 +59,13 @@ public Mono<PublishReply> publish(Mono<PublishRequest> requestMono) {
                 ))
                 .transform(mono -> {
                     for (RecordPreProcessor processor : recordPreProcessorChain.getAll()) {
-                        mono = mono.flatMap(envelope -> Mono.fromCompletionStage(processor.preProcess(envelope)));
+                        mono = mono.flatMap(envelope -> {
+                            try {
+                                return Mono.fromCompletionStage(processor.preProcess(envelope));
+                            } catch (Throwable e) {
+                                return Mono.error(new RuntimeException(processor + " failed", e));
+                            }
+                        });
                     }
                     return mono;
                 })
@@ -69,7 +75,8 @@ public Mono<PublishReply> publish(Mono<PublishRequest> requestMono) {
                         .setPartition(it.getPartition())
                         .setOffset(it.getOffset())
                         .build()
-                );
+                )
+                .log("publish", Level.SEVERE, SignalType.ON_ERROR);
     }
 
     @Override
@@ -124,7 +131,7 @@ public Flux<SubscribeReply> subscribe(Mono<SubscribeRequest> requestFlux) {
                                 subscriptions.remove(sessionId, subscription);
                             });
                 })
-                .log("subscribe", Level.WARNING, SignalType.ON_ERROR);
+                .log("subscribe", Level.SEVERE, SignalType.ON_ERROR);
     }
 
     @Override
@@ -162,7 +169,7 @@ public Flux<ReceiveReply> receive(Mono<ReceiveRequest> requestMono) {
                                     .build()
                             );
                 })
-                .log("receive", Level.WARNING, SignalType.ON_ERROR);
+                .log("receive", Level.SEVERE, SignalType.ON_ERROR);
     }
 
     @Override
@@ -184,7 +191,7 @@ public Mono<Empty> ack(Mono<AckRequest> request) {
                     ));
                 })
                 .then(Mono.just(Empty.getDefaultInstance()))
-                .log("ack", Level.WARNING, SignalType.ON_ERROR);
+                .log("ack", Level.SEVERE, SignalType.ON_ERROR);
     }
 
     @Override