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.component.ignite.cache;
  
  import java.util.Map;
  
  import javax.cache.Cache.Entry;
  
  import org.apache.camel.CamelException;
  import org.apache.camel.Category;
  import org.apache.camel.Consumer;
  import org.apache.camel.Processor;
  import org.apache.camel.Producer;
  import org.apache.camel.component.ignite.AbstractIgniteEndpoint;
  import org.apache.camel.component.ignite.IgniteConstants;
  import org.apache.camel.spi.Metadata;
  import org.apache.camel.spi.UriEndpoint;
  import org.apache.camel.spi.UriParam;
  import org.apache.camel.spi.UriPath;
  import org.apache.ignite.IgniteCache;
  import org.apache.ignite.cache.CacheEntryEventSerializableFilter;
  import org.apache.ignite.cache.CachePeekMode;
  import org.apache.ignite.cache.query.ContinuousQuery;
  import org.apache.ignite.cache.query.Query;
  
  import static org.apache.camel.component.ignite.IgniteConstants.SCHEME_CACHE;
  
  /**
   * Perform cache operations on an Ignite cache or consume changes from a continuous query.
   */
  @UriEndpoint(firstVersion = "2.17.0", scheme = SCHEME_CACHE, title = "Ignite Cache", syntax = "ignite-cache:cacheName",
               category = { Category.CACHE, Category.CLUSTERING }, headersClass = IgniteConstants.class)
  public class IgniteCacheEndpoint extends AbstractIgniteEndpoint {
  
      @UriPath
      @Metadata(required = true)
      private String cacheName;
  
      @UriParam(label = "producer")
      private IgniteCacheOperation operation;
  
      @UriParam(label = "producer", defaultValue = "false")
      private boolean failIfInexistentCache;
  
      @UriParam(label = "producer", defaultValue = "ALL")
      private CachePeekMode cachePeekMode = CachePeekMode.ALL;
  
      @UriParam(label = "producer,consumer")
      private Query<Entry<Object, Object>> query;
  
      @UriParam(label = "consumer")
      private CacheEntryEventSerializableFilter<Object, Object> remoteFilter;
  
      @UriParam(label = "consumer", defaultValue = "true")
      private boolean oneExchangePerUpdate = true;
  
      @UriParam(label = "consumer", defaultValue = "false")
      private boolean fireExistingQueryResults;
  
      @UriParam(label = "consumer", defaultValue = "true", defaultValueNote = "ContinuousQuery.DFLT_AUTO_UNSUBSCRIBE")
      private boolean autoUnsubscribe = ContinuousQuery.DFLT_AUTO_UNSUBSCRIBE;
  
      @UriParam(label = "consumer", defaultValue = "1", defaultValueNote = "ContinuousQuery.DFLT_PAGE_SIZE")
      private int pageSize = ContinuousQuery.DFLT_PAGE_SIZE;
  
      @UriParam(label = "consumer", defaultValue = "0", defaultValueNote = "ContinuousQuery.DFLT_TIME_INTERVAL")
      private long timeInterval = ContinuousQuery.DFLT_TIME_INTERVAL;
  
      public IgniteCacheEndpoint(String endpointUri, String remaining, Map<String, Object> parameters,
                                 IgniteCacheComponent igniteComponent) {
          super(endpointUri, igniteComponent);
          cacheName = remaining;
      }
  
      @Override
      public Producer createProducer() throws Exception {
          return new IgniteCacheProducer(this, obtainCache());
      }
  
      @Override
      public Consumer createConsumer(Processor processor) throws Exception {
          Consumer consumer = new IgniteCacheContinuousQueryConsumer(this, processor);
          configureConsumer(consumer);
          return consumer;
      }
  
      protected IgniteCache<Object, Object> obtainCache() throws CamelException {
          IgniteCache<Object, Object> cache = ignite().cache(cacheName);
          if (cache == null) {
              if (failIfInexistentCache) {
                  throw new CamelException(
                          String.format("Ignite cache %s doesn't exist, and failIfInexistentCache is true", cacheName));
              }
              cache = ignite().createCache(cacheName);
          }
  
          return cache;
      }
  
      /**
       * Gets the cache name.
       *
       * @return
       */
      public String getCacheName() {
          return cacheName;
      }
  
      /**
       * The cache name.
       *
       * @param cacheName cache name
       */
      public void setCacheName(String cacheName) {
          this.cacheName = cacheName;
      }
  
      /**
       * Gets the cache operation to invoke.
       *
       * @return cache name
       */
      public IgniteCacheOperation getOperation() {
          return operation;
      }
  
      /**
       * The cache operation to invoke.
       * <p>
       * Possible values: GET, PUT, REMOVE, SIZE, REBALANCE, QUERY, CLEAR.
       * </p>
       *
       * @param operation
       */
      public void setOperation(IgniteCacheOperation operation) {
          this.operation = operation;
      }
  
      /**
       * Whether to fail the initialization if the cache doesn't exist.
       *
       * @return
       */
      public boolean isFailIfInexistentCache() {
          return failIfInexistentCache;
      }
  
      /**
       * Whether to fail the initialization if the cache doesn't exist.
       *
       * @param failIfInexistentCache
       */
      public void setFailIfInexistentCache(boolean failIfInexistentCache) {
          this.failIfInexistentCache = failIfInexistentCache;
      }
  
      /**
       * Gets the {@link CachePeekMode}, only needed for operations that require it ({@link IgniteCacheOperation#SIZE}).
       *
       * @return
       */
      public CachePeekMode getCachePeekMode() {
          return cachePeekMode;
      }
  
      /**
       * The {@link CachePeekMode}, only needed for operations that require it ({@link IgniteCacheOperation#SIZE}).
       *
       * @param cachePeekMode
       */
      public void setCachePeekMode(CachePeekMode cachePeekMode) {
          this.cachePeekMode = cachePeekMode;
      }
  
      /**
       * Gets the query to execute, only needed for operations that require it, and for the Continuous Query Consumer.
       *
       * @return
       */
      public Query<Entry<Object, Object>> getQuery() {
          return query;
      }
  
      /**
       * The {@link Query} to execute, only needed for operations that require it, and for the Continuous Query Consumer.
       *
       * @param query
       */
      public void setQuery(Query<Entry<Object, Object>> query) {
          this.query = query;
      }
  
      /**
       * Gets the remote filter, only used by the Continuous Query Consumer.
       *
       * @return
       */
      public CacheEntryEventSerializableFilter<Object, Object> getRemoteFilter() {
          return remoteFilter;
      }
  
      /**
       * The remote filter, only used by the Continuous Query Consumer.
       *
       * @param remoteFilter
       */
      public void setRemoteFilter(CacheEntryEventSerializableFilter<Object, Object> remoteFilter) {
          this.remoteFilter = remoteFilter;
      }
  
      /**
       * Gets whether to pack each update in an individual Exchange, even if multiple updates are received in one batch.
       * Only used by the Continuous Query Consumer.
       *
       * @return
       */
      public boolean isOneExchangePerUpdate() {
          return oneExchangePerUpdate;
      }
  
      /**
       * Whether to pack each update in an individual Exchange, even if multiple updates are received in one batch. Only
       * used by the Continuous Query Consumer.
       *
       * @param oneExchangePerUpdate
       */
      public void setOneExchangePerUpdate(boolean oneExchangePerUpdate) {
          this.oneExchangePerUpdate = oneExchangePerUpdate;
      }
  
      /**
       * Gets whether auto unsubscribe is enabled in the Continuous Query Consumer.
       *
       * @return
       */
      public boolean isAutoUnsubscribe() {
          return autoUnsubscribe;
      }
  
      /**
       * Whether auto unsubscribe is enabled in the Continuous Query Consumer.
       *
       * @param autoUnsubscribe
       */
      public void setAutoUnsubscribe(boolean autoUnsubscribe) {
          this.autoUnsubscribe = autoUnsubscribe;
      }
  
      /**
       * Gets the page size. Only used by the Continuous Query Consumer.
       *
       * @return
       */
      public int getPageSize() {
          return pageSize;
      }
  
      /**
       * The page size. Only used by the Continuous Query Consumer.
       *
       * @param pageSize
       */
      public void setPageSize(int pageSize) {
          this.pageSize = pageSize;
      }
  
      /**
       * Gets whether to process existing results that match the query. Used on initialization of the Continuous Query
       * Consumer.
       *
       * @return
       */
      public boolean isFireExistingQueryResults() {
          return fireExistingQueryResults;
      }
  
      /**
       * Whether to process existing results that match the query. Used on initialization of the Continuous Query
       * Consumer.
       *
       * @param fireExistingQueryResults
       */
      public void setFireExistingQueryResults(boolean fireExistingQueryResults) {
          this.fireExistingQueryResults = fireExistingQueryResults;
      }
  
      /**
       * Gets the time interval for the Continuous Query Consumer.
       *
       * @return
       */
      public long getTimeInterval() {
          return timeInterval;
      }
  
      /**
       * The time interval for the Continuous Query Consumer.
       *
       * @param timeInterval
       */
      public void setTimeInterval(long timeInterval) {
          this.timeInterval = timeInterval;
      }
  
  }
  