Send kafka msg to ingest #164

pom.xml

@@ -73,6 +73,12 @@
             <scope>test</scope>
         </dependency>
 
+        <!-- Kafka -->
+        <dependency>
+            <groupId>org.springframework.kafka</groupId>
+            <artifactId>spring-kafka</artifactId>
+        </dependency>
+
         <!-- Graphql java dependencies -->
         <dependency>
             <groupId>com.graphql-java</groupId>

src/main/java/bio/overture/songsearch/config/kafka/AnalysisMessage.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright (c) 2018. Ontario Institute for Cancer Research
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package bio.overture.songsearch.config.kafka;
+
+import static lombok.AccessLevel.PRIVATE;
+import bio.overture.songsearch.model.enums.AnalysisState;
+import lombok.*;
+
+@Value
+// Note: although the AllArgs and NoArgs combination below seems odd,
+// it allows Jackson to deserialize to an immutable object without using any additional annotations.
+@AllArgsConstructor
+@NoArgsConstructor(force = true, access = PRIVATE)
+public class AnalysisMessage {
+
+  @NonNull private final String analysisId;
+  @NonNull private final String studyId;
+  @NonNull private final String state;
+  @NonNull private final String songServerId;
+  @NonNull private final Analysis analysis;
+
+  public static AnalysisMessage createAnalysisMessage(
+      bio.overture.songsearch.model.Analysis analysis, String songServerId) {
+
+    return new AnalysisMessage(
+        analysis.getAnalysisId(),
+        analysis.getStudyId(),
+        analysis.getAnalysisState().toString(),
+        analysis.getRepositories().get(0).getCode(),
+        new Analysis(
+            analysis.getAnalysisId(),
+            new AnalysisType(analysis.getAnalysisType()),
+            analysis.getAnalysisState(),
+            analysis.getStudyId()));
+  }
+
+  @Value
+  @AllArgsConstructor
+  private static class Analysis {
+    String analysisId;
+    AnalysisType analysisType;
+    AnalysisState analysisState;
+    String studyId;
+  }
+
+  @Data
+  @AllArgsConstructor
+  public static class AnalysisType {
+    String name;
+  }
+}

src/main/java/bio/overture/songsearch/config/kafka/DefaultSender.java

@@ -0,0 +1,13 @@
+package bio.overture.songsearch.config.kafka;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+public class DefaultSender implements Sender {
+
+  public void send(String payload, String key) {
+    log.debug("key: " + key);
+  }
+}

src/main/java/bio/overture/songsearch/config/kafka/KafkaConfig.java

@@ -0,0 +1,53 @@
+package bio.overture.songsearch.config.kafka;
+
+import java.util.HashMap;
+import java.util.Map;
+import lombok.val;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Profile;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+
+@Configuration
+@Profile("kafka")
+public class KafkaConfig {
+
+  @Value("${spring.kafka.bootstrap-servers}")
+  private String bootstrapServers;
+
+  @Value("${spring.kafka.template.default-topic}")
+  private String defaultTopic;
+
+  @Bean
+  public Map<String, Object> producerConfigs() {
+    val props = new HashMap<String, Object>();
+    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+
+    return props;
+  }
+
+  @Bean
+  public ProducerFactory<String, String> producerFactory() {
+    return new DefaultKafkaProducerFactory<>(producerConfigs());
+  }
+
+  @Bean
+  public KafkaTemplate<String, String> kafkaTemplate() {
+    val template = new KafkaTemplate<>(producerFactory());
+    template.setDefaultTopic(defaultTopic);
+    System.out.println("defaultTopic: " + defaultTopic);
+    return template;
+  }
+
+  @Bean
+  public KafkaSender sender() {
+    return new KafkaSender();
+  }
+}

src/main/java/bio/overture/songsearch/config/kafka/KafkaSender.java

@@ -0,0 +1,17 @@
+package bio.overture.songsearch.config.kafka;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Profile;
+import org.springframework.kafka.core.KafkaTemplate;
+
+@Slf4j
+@Profile("kafka")
+public class KafkaSender implements Sender {
+  @Autowired private KafkaTemplate<String, String> kafkaTemplate;
+
+  public void send(String payload, String key) {
+    log.debug("sending payload='{}' to topic='{}'", payload, kafkaTemplate.getDefaultTopic());
+    kafkaTemplate.send(kafkaTemplate.getDefaultTopic(), key, payload);
+  }
+}

src/main/java/bio/overture/songsearch/config/kafka/Sender.java

@@ -0,0 +1,5 @@
+package bio.overture.songsearch.config.kafka;
+
+public interface Sender {
+  void send(String payload, String key);
+}

src/main/java/bio/overture/songsearch/graphql/GraphQLProvider.java

@@ -52,6 +52,8 @@ public class GraphQLProvider {
   private final AnalysisDataFetcher analysisDataFetcher;
   private final FileDataFetcher fileDataFetcher;
   private final EntityDataFetcher entityDataFetcher;
+
+  private final StartAutomationMutation startAutomationMutation;
   private final AuthProperties authProperties;
   private GraphQL graphQL;
   private GraphQLSchema graphQLSchema;
@@ -61,10 +63,12 @@ public GraphQLProvider(
       AnalysisDataFetcher analysisDataFetcher,
       FileDataFetcher fileDataFetcher,
       EntityDataFetcher entityDataFetcher,
+      StartAutomationMutation startAutomationMutation,
       AuthProperties authProperties) {
     this.analysisDataFetcher = analysisDataFetcher;
     this.fileDataFetcher = fileDataFetcher;
     this.entityDataFetcher = entityDataFetcher;
+    this.startAutomationMutation = startAutomationMutation;
     this.authProperties = authProperties;
   }
 
@@ -147,6 +151,9 @@ private RuntimeWiring buildWiring() {
                 .dataFetcher(
                     "sampleMatchedAnalysesForDonor",
                     analysisDataFetcher.getSampleMatchedAnalysesForDonorFetcher()))
+        .type(
+            newTypeWiring("Mutation")
+                .dataFetcher("startAutomation", startAutomationMutation.startAutomationResolver()))
         .build();
   }
 

src/main/java/bio/overture/songsearch/graphql/StartAutomationMutation.java

@@ -0,0 +1,97 @@
+/*
+ * Copyright (c) 2021 The Ontario Institute for Cancer Research. All rights reserved
+ *
+ * This program and the accompanying materials are made available under the terms of the GNU Affero General Public License v3.0.
+ * You should have received a copy of the GNU Affero General Public License along with
+ * this program. If not, see <http://www.gnu.org/licenses/>.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY
+ * EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
+ * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT
+ * SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
+ * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED
+ * TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS;
+ * OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER
+ * IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
+ * ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package bio.overture.songsearch.graphql;
+
+import static bio.overture.songsearch.config.kafka.AnalysisMessage.createAnalysisMessage;
+import static bio.overture.songsearch.utils.JacksonUtils.convertValue;
+import static java.util.stream.Collectors.toUnmodifiableList;
+
+import bio.overture.songsearch.config.kafka.Sender;
+import bio.overture.songsearch.model.Analysis;
+import bio.overture.songsearch.model.Sort;
+import bio.overture.songsearch.service.AnalysisService;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import graphql.schema.DataFetcher;
+import java.util.List;
+import java.util.Map;
+
+import lombok.NonNull;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+import lombok.val;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+
+@Component
+@Slf4j
+public class StartAutomationMutation {
+
+  private final AnalysisService analysisService;
+
+  @Value("${songServerId")
+  private String songServerId;
+
+  private final Sender sender;
+
+  @Autowired
+  public StartAutomationMutation(AnalysisService analysisService, @NonNull Sender sender) {
+    this.analysisService = analysisService;
+    this.sender = sender;
+  }
+
+  public DataFetcher<Analysis> startAutomationResolver() {
+    return env -> {
+      val args = env.getArguments();
+
+      val filter = ImmutableMap.<String, Object>builder();
+      val page = ImmutableMap.<String, Integer>builder();
+      val sorts = ImmutableList.<Sort>builder();
+
+      if (args != null) {
+        if (args.get("filter") != null) filter.putAll((Map<String, Object>) args.get("filter"));
+        if (args.get("page") != null) page.putAll((Map<String, Integer>) args.get("page"));
+        if (args.get("sorts") != null) {
+          val rawSorts = (List<Object>) args.get("sorts");
+          sorts.addAll(
+              rawSorts.stream()
+                  .map(sort -> convertValue(sort, Sort.class))
+                  .collect(toUnmodifiableList()));
+        }
+      }
+
+      Analysis analysis =
+          analysisService.getAnalysisById(env.getArguments().get("analysisId").toString());
+      log.debug("Analysis fetched: " + analysis);
+
+      sendAnalysisMessage(analysis);
+      log.debug("Message sent to kafka queue");
+      return analysis;
+    };
+  }
+
+  @SneakyThrows
+  public void sendAnalysisMessage(Analysis analysis) {
+    val message = createAnalysisMessage(analysis, songServerId);
+    log.debug("Message payload:: " + new ObjectMapper().writeValueAsString(message));
+    sender.send(new ObjectMapper().writeValueAsString(message), message.getAnalysisId());
+  }
+}

src/main/java/bio/overture/songsearch/repository/AnalysisRepository.java

@@ -84,7 +84,9 @@ private static Map<String, Function<Object, AbstractQueryBuilder<?>>> argumentPa
             REPOSITORY_CODE,
             value ->
                 new NestedQueryBuilder(
-                    "repositories", new TermQueryBuilder("repositories.code", value), ScoreMode.None))
+                    "repositories",
+                    new TermQueryBuilder("repositories.code", value),
+                    ScoreMode.None))
         .put(
             DONOR_ID,
             value ->

src/main/java/bio/overture/songsearch/service/AnalysisService.java

@@ -35,19 +35,22 @@
 import lombok.NonNull;
 import lombok.SneakyThrows;
 import lombok.Value;
+import lombok.extern.slf4j.Slf4j;
 import lombok.val;
 import org.elasticsearch.action.search.MultiSearchResponse;
 import org.elasticsearch.search.SearchHit;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
 @Service
+@Slf4j
 public class AnalysisService {
 
   private final AnalysisRepository analysisRepository;
 
   @Autowired
   public AnalysisService(AnalysisRepository analysisRepository) {
+
     this.analysisRepository = analysisRepository;
   }
 

src/main/resources/application.yaml

@@ -34,3 +34,15 @@ auth:
         - RDPC-DEV.READ
       queryAndMutation:
         - RDPC-DEV.WRITE
+---
+
+spring:
+  config:
+    activate:
+      on-profile: kafka
+  kafka:
+    bootstrap-servers: localhost:9092
+    template:
+      default-topic: automation_trigger #song-analysis
+
+songServerId: submission-song.collab

src/main/resources/schema.graphql

@@ -251,3 +251,11 @@ extend type Query {
     """
     sampleMatchedAnalysesForDonor(req: SampleMatchedAnalysesForDonorReq!): [SampleMatchedAnalysisPair]
 }
+
+type Mutation {
+    startAutomation(analysisId: String!): MutationResponse
+}
+
+type MutationResponse {
+    analysisId: ID
+}