Call reply closure on method event subscriptions

grails-events/src/main/groovy/org/grails/events/EventSubscriberTrigger.groovy

@@ -18,24 +18,36 @@ import groovy.util.logging.Slf4j
 class EventSubscriberTrigger implements EventTrigger {
     final Event event
     final Subscriber subscriber
+    final Closure reply
 
-    EventSubscriberTrigger(Event event, Subscriber subscriber) {
+    EventSubscriberTrigger(Event event, Subscriber subscriber, Closure reply = null) {
         this.event = event
         this.subscriber = subscriber
+        this.reply = reply
     }
 
     @Override
     Object proceed() {
         try {
+            def result
             if(subscriber instanceof EventSubscriber) {
-                return subscriber.call(event)
+                result = subscriber.call(event)
             }
             else {
-                return subscriber.call(event.data)
+                result = subscriber.call(event.data)
             }
+            if(reply != null) {
+                return reply.call(result)
+            }
+            return result
         } catch (Throwable e) {
-            log.error("Error triggering event [$event.id] for subscriber [${subscriber}]: $e.message", e)
-            throw e
+            if(reply != null && reply.parameterTypes && reply.parameterTypes[0].isInstance(e)) {
+                reply.call(e)
+            }
+            else {
+                log.error("Error triggering event [$event.id] for subscriber [${subscriber}]: $e.message", e)
+                throw e
+            }
         }
     }
 }

grails-events/src/main/groovy/org/grails/events/registry/EventSubscriberSubscription.groovy

@@ -29,6 +29,6 @@ class EventSubscriberSubscription extends AbstractSubscription {
 
     @Override
     EventTrigger buildTrigger(Event event, Closure reply) {
-        return new EventSubscriberTrigger(event, subscriber)
+        return new EventSubscriberTrigger(event, subscriber, reply)
     }
 }

grails-events/src/test/groovy/org/grails/events/subscriber/MethodEventSubscriberSpec.groovy

@@ -1,13 +1,22 @@
 package org.grails.events.subscriber
 
 import grails.events.subscriber.MethodSubscriber
+import org.grails.events.bus.SynchronousEventBus
+import spock.lang.Shared
 import spock.lang.Specification
+import spock.lang.Unroll
 
 /**
  * Created by graemerocher on 28/03/2017.
  */
 class MethodEventSubscriberSpec extends Specification {
 
+    @Shared TestHandler testHandler = new TestHandler()
+
+    def setup() {
+        testHandler.eventHandled = false
+    }
+
     void "test convert method argument"() {
         given:
         TestService testService = new TestService()
@@ -18,10 +27,44 @@ class MethodEventSubscriberSpec extends Specification {
         subscriber.call("1") == 2
         subscriber.call("") == null
     }
+
+    @Unroll
+    void "sendAndReceive calls reply once subscriber handles the method - #type"() {
+
+        given:
+        def eventBus = new SynchronousEventBus()
+        def topic = 'test_topic'
+        def replied = false
+
+        and:
+        eventBus.subscribe(topic, subscriber)
+
+        when:
+        eventBus.sendAndReceive(topic, new Object()) {
+            replied = true
+        }
+
+        then:
+        testHandler.eventHandled
+        replied
+
+        where:
+        type      | subscriber
+        'closure' | { testHandler.handleEvent() }
+        'method'  | new MethodSubscriber(testHandler, TestHandler.getMethod("handleEvent"))
+    }
 }
 
 class TestService {
     def foo(Integer num) {
        return num + 1
     }
 }
+
+class TestHandler {
+    boolean eventHandled = false
+
+    def handleEvent() {
+        eventHandled = true
+    }
+}