content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements.  See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License.  You may obtain a copy of the License at
   *
   *      http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.camel.example.debezium;
  
  import java.util.Arrays;
  import java.util.Collections;
  import java.util.Map;
  
  import org.apache.camel.builder.RouteBuilder;
  import org.apache.camel.main.Main;
  import org.apache.camel.model.dataformat.JsonLibrary;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  /**
   * A simple example to sink data from Kinesis that produced by Debezium into Cassandra
   */
  public final class KinesisProducerToCassandra {
  
      private static final Logger LOG = LoggerFactory.getLogger(KinesisProducerToCassandra.class);
  
      // use Camel Main to set up and run Camel
      private static final Main MAIN = new Main();
  
      private KinesisProducerToCassandra() {
      }
  
      public static void main(String[] args) throws Exception {
  
          LOG.debug("About to run Kinesis to Cassandra integration...");
  
          // add route
          MAIN.configure().addRoutesBuilder(createRouteBuilder());
  
          // start and run Camel (block)
          MAIN.run();
      }
  
      static RouteBuilder createRouteBuilder() {
          return new RouteBuilder() {
              public void configure() {
                  // We set the CQL templates we need, note that an UPDATE in Cassandra means an UPSERT which is what we need
                  final String cqlUpdate = "update products set name = ?, description = ?, weight = ? where id = ?";
                  final String cqlDelete = "delete from products where id = ?";
  
                  from("aws2-kinesis:{{kinesis.streamName}}?accessKey=RAW({{kinesis.accessKey}})"
                          + "&secretKey=RAW({{kinesis.secretKey}})"
                          + "&region={{kinesis.region}}")
                          // Since we expect the data of the body to be ByteArr, we convert it to String using Kinesis
                          // Type Converter, in order to unmarshal later from JSON to Map
                          .convertBodyTo(String.class)
                          // Unmarshal our body, it will convert it from JSON to Map
                          .unmarshal().json(JsonLibrary.Jackson)
                          // In order not to lose the operation that we set in Debezium, we set it as a property or as
                          // a header
                          .setProperty("DBOperation", simple("${body[operation]}"))
                          .choice()
                          // If we have a INSERT or UPDATE, we will need to set the body with the CQL query parameters since we are using
                          // camel-cassandraql component
                          .when(exchangeProperty("DBOperation").in("c", "u"))
                          .setBody(exchange -> {
                              final Map body = (Map) exchange.getMessage().getBody();
                              final Map value = (Map) body.get("value");
                              final Map key = (Map) body.get("key");
  
                              // We as well check for nulls
                              final String name = value.get("name") != null ? value.get("name").toString() : "";
                              final String description = value.get("description") != null ? value.get("description").toString() : "";
                              final float weight = value.get("weight") != null ? Float.parseFloat(value.get("weight").toString()) : 0;
  
                              return Arrays.asList(name, description, weight, key.get("id"));
                          })
                          // We set the appropriate query in the header so we don't run the same route twice
                          .setHeader("CQLQuery", constant(cqlUpdate))
                          // If we have a DELETE, then we just set the id as a query parameter in the body
                          .when(exchangeProperty("DBOperation").isEqualTo("d"))
                          .setBody(exchange -> {
                              final Map body = (Map) exchange.getMessage().getBody();
                              final Map key = (Map) body.get("key");
  
                              return Collections.singletonList(key.get("id"));
                          })
                          // We set the appropriate query in the header so we don't run the same route twice
                          .setHeader("CQLQuery", constant(cqlDelete))
                          .end()
                          .choice()
                          // We just make sure we ONLY handle INSERT, UPDATE and DELETE and nothing else
                          .when(exchangeProperty("DBOperation").in("c", "u", "d"))
                          // Send query to Cassandra
                          .recipientList(simple("cql:{{cassandra.node}}/{{cassandra.keyspace}}?cql=RAW(${header.CQLQuery})"))
                          .end();
              }
          };
      }
  
  }
  