Implemented eventbus and eventhandling for git, but GitWorker needs more work

Author: pvdissel

subprojects/pipe-listener/src/main/groovy/org/pipelinelabs/pipeline/listen/PipeListenService.groovy

@@ -1,18 +1,40 @@
 package org.pipelinelabs.pipeline.listen
 
+import com.google.common.eventbus.AsyncEventBus
 import com.yammer.dropwizard.Service
 import com.yammer.dropwizard.config.Bootstrap
 import com.yammer.dropwizard.config.Environment
+import org.pipelinelabs.pipeline.listen.core.DeadEventHandler
+import org.pipelinelabs.pipeline.listen.core.GitWorker
 import org.pipelinelabs.pipeline.listen.resources.GitHubWebHookResource
 
+import static java.util.concurrent.TimeUnit.SECONDS
+
 class PipeListenService extends Service<PipeListenConfiguration> {
 
     @Override
     void initialize(Bootstrap<PipeListenConfiguration> bootstrap) {
     }
 
     @Override
-    void run(PipeListenConfiguration configuration, Environment environment) throws Exception {
-        environment.addResource(new GitHubWebHookResource())
+    void run(PipeListenConfiguration config, Environment env) throws Exception {
+        def bus = createEventBus(env)
+        env.manage(new DeadEventHandler(bus))
+        env.manage(new GitWorker(bus));
+        env.addResource(new GitHubWebHookResource(bus))
+    }
+
+    private createEventBus(Environment env) {
+        final corePoolSize = 2
+        final maxPoolSize = 6
+        final keepAliveTimeInSeconds = 30
+        new AsyncEventBus(
+                env.managedExecutorService(
+                        "eventbus-worker-%s",
+                        corePoolSize,
+                        maxPoolSize,
+                        keepAliveTimeInSeconds, SECONDS
+                )
+        )
     }
 }

subprojects/pipe-listener/src/main/groovy/org/pipelinelabs/pipeline/listen/core/DeadEventHandler.groovy

@@ -0,0 +1,32 @@
+package org.pipelinelabs.pipeline.listen.core
+
+import com.google.common.eventbus.DeadEvent
+import com.google.common.eventbus.EventBus
+import com.google.common.eventbus.Subscribe
+import com.yammer.dropwizard.lifecycle.Managed
+import groovy.util.logging.Slf4j
+
+@Slf4j
+class DeadEventHandler implements Managed {
+
+    private final EventBus bus
+
+    DeadEventHandler(EventBus bus) {
+        this.bus = bus
+    }
+
+    @Override
+    void start() throws Exception {
+        bus.register(this)
+    }
+
+    @Override
+    void stop() throws Exception {
+        bus.unregister(this)
+    }
+
+    @Subscribe
+    void work(DeadEvent event) {
+        log.error("No event handler available for event [{}]", event.event)
+    }
+}

subprojects/pipe-listener/src/main/groovy/org/pipelinelabs/pipeline/listen/core/GitTriggerEvent.groovy

@@ -0,0 +1,11 @@
+package org.pipelinelabs.pipeline.listen.core
+
+
+class GitTriggerEvent {
+
+    final String url
+
+    GitTriggerEvent(String url) {
+        this.url = url
+    }
+}

subprojects/pipe-listener/src/main/groovy/org/pipelinelabs/pipeline/listen/core/GitWorker.groovy

@@ -0,0 +1,34 @@
+package org.pipelinelabs.pipeline.listen.core
+
+import com.google.common.eventbus.EventBus
+import com.google.common.eventbus.Subscribe
+import com.yammer.dropwizard.lifecycle.Managed
+
+import java.nio.file.Files
+
+class GitWorker implements Managed {
+
+    private final EventBus bus
+
+    GitWorker(EventBus bus) {
+        this.bus = bus
+    }
+
+    @Override
+    void start() throws Exception {
+        bus.register(this)
+    }
+
+    @Override
+    void stop() throws Exception {
+        bus.unregister(this)
+    }
+
+    @Subscribe
+    void work(GitTriggerEvent event) {
+        def dir = Files.createTempDirectory("pipeline/")
+        List envProps = null
+        "git clone ${event.url}".execute(envProps, dir.toFile())
+        // TODO: Call pipe-runner
+    }
+}

subprojects/pipe-listener/src/main/groovy/org/pipelinelabs/pipeline/listen/resources/GitHubWebHookResource.groovy

@@ -1,7 +1,9 @@
 package org.pipelinelabs.pipeline.listen.resources
 
+import com.google.common.eventbus.EventBus
 import groovy.json.JsonSlurper
 import groovy.util.logging.Slf4j
+import org.pipelinelabs.pipeline.listen.core.GitTriggerEvent
 
 import javax.ws.rs.*
 import javax.ws.rs.core.Response
@@ -15,28 +17,44 @@ import static javax.ws.rs.core.Response.Status.BAD_REQUEST
 @Consumes(APPLICATION_JSON)
 class GitHubWebHookResource {
 
+    private final EventBus bus
+
+    GitHubWebHookResource(EventBus bus) {
+        this.bus = bus
+    }
+
     @POST
     Response trigger(@FormParam('payload') String payload) {
         final slurper = new JsonSlurper()
         def info = slurper.parseText(payload)
         try {
-            assert info.commits
-            assert info.head_commit
-            assert info.pusher
-            assert info.pusher.email
-            assert info.pusher.name
-            assert info.ref
-            assert info.repository
-            assert info.repository.name
-            assert info.repository.master_branch
-            assert info.repository.language
-            assert info.repository.private == true ||
-                    info.repository.private == false
-            assert info.repository.url
+            verifyRequest(info)
         } catch (AssertionError e) {
             log.warn("Received GitHub WebHook payload with unexpected format:\n{}", info)
             return Response.status(BAD_REQUEST.statusCode).build()
         }
-        return Response.noContent().build()
+        return handleRequest(info)
+    }
+
+    private Response handleRequest(request) {
+        def event = new GitTriggerEvent(request.repository.url)
+        bus.post(event)
+        Response.noContent().build()
+    }
+
+    private verifyRequest(request) throws AssertionError {
+        assert request.commits
+        assert request.head_commit
+        assert request.pusher
+        assert request.pusher.email
+        assert request.pusher.name
+        assert request.ref
+        assert request.repository
+        assert request.repository.name
+        assert request.repository.master_branch
+        assert request.repository.language
+        assert request.repository.private == true ||
+                request.repository.private == false
+        assert request.repository.url
     }
 }

subprojects/pipe-listener/src/test/groovy/org/pipelinelabs/pipeline/listen/PipeListenServiceSpec.groovy

@@ -1,21 +1,57 @@
 package org.pipelinelabs.pipeline.listen
 
 import com.yammer.dropwizard.config.Environment
+import org.pipelinelabs.pipeline.listen.core.DeadEventHandler
+import org.pipelinelabs.pipeline.listen.core.GitWorker
 import org.pipelinelabs.pipeline.listen.resources.GitHubWebHookResource
 import spock.lang.Specification
 
+import java.util.concurrent.ExecutorService
+
 class PipeListenServiceSpec extends Specification {
 
     def 'Service serves the GitHubWebHookResource'() {
         given:
         def env = Mock(Environment)
+        def executor = Mock(ExecutorService)
         def service = new PipeListenService()
         def config = new PipeListenConfiguration()
 
         when:
         service.run(config, env)
 
         then:
+        1 * env.managedExecutorService(_, _, _, _, _) >> executor
         1 * env.addResource(_ as GitHubWebHookResource)
     }
+
+    def 'Service manages the GitWorker'() {
+        given:
+        def env = Mock(Environment)
+        def executor = Mock(ExecutorService)
+        def service = new PipeListenService()
+        def config = new PipeListenConfiguration()
+
+        when:
+        service.run(config, env)
+
+        then:
+        1 * env.managedExecutorService(_, _, _, _, _) >> executor
+        1 * env.manage(_ as GitWorker)
+    }
+
+    def 'Service manages the DeadEventHandler'() {
+        given:
+        def env = Mock(Environment)
+        def executor = Mock(ExecutorService)
+        def service = new PipeListenService()
+        def config = new PipeListenConfiguration()
+
+        when:
+        service.run(config, env)
+
+        then:
+        1 * env.managedExecutorService(_, _, _, _, _) >> executor
+        1 * env.manage(_ as DeadEventHandler)
+    }
 }

subprojects/pipe-listener/src/test/groovy/org/pipelinelabs/pipeline/listen/core/DeadEventHandlerSpec.groovy

@@ -0,0 +1,50 @@
+package org.pipelinelabs.pipeline.listen.core
+
+import com.google.common.eventbus.DeadEvent
+import com.google.common.eventbus.EventBus
+import com.google.common.eventbus.Subscribe
+import spock.lang.Ignore
+import spock.lang.Specification
+
+
+class DeadEventHandlerSpec extends Specification {
+    private EventBus bus = Mock(EventBus)
+
+    def 'Start registers itself to the event bus'() {
+        given:
+        def worker = new DeadEventHandler(bus)
+
+        when:
+        worker.start()
+
+        then:
+        1 * bus.register(worker)
+    }
+
+    def 'Stop unregisters itself from the event bus'() {
+        given:
+        def worker = new DeadEventHandler(bus)
+
+        when:
+        worker.stop()
+
+        then:
+        1 * bus.unregister(worker)
+    }
+
+    def 'Work handles DeadEvent and is annotated with @Subscribe'() {
+        def method = DeadEventHandler.getMethod("work", DeadEvent)
+
+        expect:
+        method.isAnnotationPresent(Subscribe)
+    }
+
+    @Ignore('How to verify if injected logger is used?')
+    def 'Work logs an error'() {
+        def worker = new DeadEventHandler(bus)
+        final event = Mock(DeadEvent)
+
+        expect:
+        worker.work(event)
+    }
+}

subprojects/pipe-listener/src/test/groovy/org/pipelinelabs/pipeline/listen/core/GitTriggerEventSpec.groovy

@@ -0,0 +1,13 @@
+package org.pipelinelabs.pipeline.listen.core
+
+import spock.lang.Specification
+
+class GitTriggerEventSpec extends Specification {
+
+    def 'Event has an url'() {
+        def url = "url"
+
+        expect:
+        new GitTriggerEvent(url).url == url
+    }
+}

subprojects/pipe-listener/src/test/groovy/org/pipelinelabs/pipeline/listen/core/GitWorkerSpec.groovy

@@ -0,0 +1,48 @@
+package org.pipelinelabs.pipeline.listen.core
+
+import com.google.common.eventbus.EventBus
+import com.google.common.eventbus.Subscribe
+import spock.lang.Ignore
+import spock.lang.Specification
+
+class GitWorkerSpec extends Specification {
+    private EventBus bus = Mock(EventBus)
+
+    def 'Start registers itself to the event bus'() {
+        given:
+        def worker = new GitWorker(bus)
+
+        when:
+        worker.start()
+
+        then:
+        1 * bus.register(worker)
+    }
+
+    def 'Stop unregisters itself from the event bus'() {
+        given:
+        def worker = new GitWorker(bus)
+
+        when:
+        worker.stop()
+
+        then:
+        1 * bus.unregister(worker)
+    }
+
+    def 'Work handles GitTriggerEvent and is annotated with @Subscribe'() {
+        def method = GitWorker.getMethod("work", GitTriggerEvent)
+
+        expect:
+        method.isAnnotationPresent(Subscribe)
+    }
+
+    @Ignore('Needs to be testable somehow')
+    def 'Work clones the git repo and starts pipe-runner'() {
+        def worker = new GitWorker(bus)
+        final event = Mock(GitTriggerEvent)
+
+        expect:
+        worker.work(event)
+    }
+}