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.quarkus.component.cassandraql.it;
  
  import com.datastax.oss.quarkus.runtime.api.session.QuarkusCqlSession;
  import jakarta.enterprise.context.ApplicationScoped;
  import jakarta.inject.Inject;
  import org.apache.camel.AggregationStrategy;
  import org.apache.camel.BindToRegistry;
  import org.apache.camel.builder.RouteBuilder;
  import org.apache.camel.processor.aggregate.cassandra.CassandraAggregationRepository;
  import org.apache.camel.processor.aggregate.cassandra.NamedCassandraAggregationRepository;
  import org.apache.camel.processor.idempotent.cassandra.NamedCassandraIdempotentRepository;
  import org.apache.camel.spi.AggregationRepository;
  import org.eclipse.microprofile.config.inject.ConfigProperty;
  
  @ApplicationScoped
  public class CassandraqlRoutes extends RouteBuilder {
      public static final String KEYSPACE = "test";
  
      @ConfigProperty(name = "quarkus.cassandra.contact-points")
      String dbUrl;
  
      @ConfigProperty(name = "quarkus.cassandra.auth.username")
      String userName;
  
      @ConfigProperty(name = "quarkus.cassandra.auth.password")
      String password;
  
      @Inject
      @BindToRegistry("quarkusCqlSession")
      QuarkusCqlSession session;
  
      @Override
      public void configure() throws Exception {
          from("direct:create")
                  .toF("cql://%s/%s?username=%s&password=%s&cql=INSERT INTO employee (id, name, address) VALUES (?, ?, ?)", dbUrl,
                          KEYSPACE, userName, password);
  
          from("direct:createIdempotent")
                  .idempotentConsumer(simple("${body[0]}"), new NamedCassandraIdempotentRepository(session, "ID"))
                  .toF("cql://%s/%s?username=%s&password=%s&cql=INSERT INTO employee (id, name, address) VALUES (?, ?, ?)", dbUrl,
                          KEYSPACE, userName, password);
  
          from("direct:createCustomSession")
                  .toF("cql:bean:customCqlSession?cql=INSERT INTO employee (id, name, address) VALUES (?, ?, ?)");
  
          from("direct:createQuarkusSession")
                  .to("cql:bean:quarkusCqlSession?cql=INSERT INTO employee (id, name, address) VALUES (?, ?, ?)");
  
          from("direct:createCustomLoadBalancingPolicy")
                  .toF("cql://%s/%s?username=%s&password=%s&loadBalancingPolicyClass=%s&cql=INSERT INTO employee (id, name, address) VALUES (?, ?, ?)",
                          dbUrl, KEYSPACE, userName, password, CustomLoadBalancingPolicy.class.getName());
  
          from("direct:read")
                  .toF("cql://%s/%s?username=%s&password=%s&cql=SELECT * FROM employee WHERE id = ?", dbUrl, KEYSPACE, userName,
                          password);
  
          from("direct:update")
                  .toF("cql://%s/%s?username=%s&password=%s&cql=UPDATE employee SET name = ?, address = ? WHERE id = ?", dbUrl,
                          KEYSPACE, userName, password);
  
          from("direct:delete")
                  .toF("cql://%s/%s?username=%s&password=%s&cql=DELETE FROM employee WHERE id = ?", dbUrl, KEYSPACE, userName,
                          password);
  
          fromF("cql://%s/%s?username=%s&password=%s&repeatCount=1&cql=SELECT * FROM employee", dbUrl, KEYSPACE, userName,
                  password).id("employee-consumer")
                  .autoStartup(false)
                  .to("seda:employees");
  
          from("direct:aggregate")
                  .aggregate(simple("${body.id}"), createAggregationStrategy())
                  .completionSize(3)
                  .completionTimeout(5000)
                  .aggregationRepository(createAggregationRepository())
                  .to("seda:employees");
  
          from("direct:readWithCustomStrategy")
                  .toF("cql://%s/%s?username=%s&password=%s&resultSetConversionStrategy=#customResultSetConversionStrategy&cql=SELECT * FROM employee WHERE id = ?",
                          dbUrl, KEYSPACE, userName, password);
  
          from("direct:cqlHeaderQuery")
                  .toF("cql://%s/%s?username=%s&password=%s", dbUrl, KEYSPACE, userName, password);
      }
  
      private AggregationStrategy createAggregationStrategy() {
          return (oldExchange, newExchange) -> {
              if (oldExchange == null) {
                  return newExchange;
              }
              Employee newBody = newExchange.getMessage().getBody(Employee.class);
              Object oldBody = oldExchange.getMessage().getBody();
  
              String newName = null;
              if (oldBody instanceof Employee) {
                  newName = ((Employee) oldBody).getName() + "," + newBody.getName();
              } else if (oldBody instanceof String) {
                  newName = oldBody + "," + newBody.getName();
              }
  
              oldExchange.getMessage().setBody(newName);
              return oldExchange;
          };
      }
  
      private AggregationRepository createAggregationRepository() {
          CassandraAggregationRepository repository = new NamedCassandraAggregationRepository();
          repository.setSession(session);
          return repository;
      }
  }
  