<commit_before>/*
 * Licensed 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.gbif.occurrence.download.inject;

import org.gbif.api.model.event.Event;
import org.gbif.api.model.occurrence.DownloadFormat;
import org.gbif.api.model.occurrence.Occurrence;
import org.gbif.dwc.terms.DwcTerm;
import org.gbif.event.search.es.SearchHitEventConverter;
import org.gbif.occurrence.download.conf.WorkflowConfiguration;
import org.gbif.occurrence.download.file.DownloadAggregator;
import org.gbif.occurrence.download.file.DownloadJobConfiguration;
import org.gbif.occurrence.download.file.DownloadMaster;
import org.gbif.occurrence.download.file.OccurrenceMapReader;
import org.gbif.occurrence.download.file.dwca.DwcaDownloadAggregator;
import org.gbif.occurrence.download.file.simplecsv.SimpleCsvDownloadAggregator;
import org.gbif.occurrence.download.file.specieslist.SpeciesListDownloadAggregator;
import org.gbif.occurrence.download.oozie.DownloadPrepareAction;
import org.gbif.occurrence.search.es.EsConfig;
import org.gbif.occurrence.search.es.EsFieldMapper;
import org.gbif.occurrence.search.es.SearchHitConverter;
import org.gbif.occurrence.search.es.SearchHitOccurrenceConverter;
import org.gbif.registry.ws.client.OccurrenceDownloadClient;
import org.gbif.wrangler.lock.Mutex;
import org.gbif.wrangler.lock.ReadWriteMutexFactory;
import org.gbif.wrangler.lock.zookeeper.ZookeeperSharedReadWriteMutex;
import org.gbif.ws.client.ClientBuilder;
import org.gbif.ws.json.JacksonJsonObjectMapperProvider;

import java.io.IOException;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.Map;
import java.util.function.Function;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.http.HttpHost;
import org.elasticsearch.client.NodeSelector;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.sniff.SniffOnFailureListener;
import org.elasticsearch.client.sniff.Sniffer;

import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import lombok.Builder;
import lombok.Data;
import lombok.experimental.UtilityClass;

;

/**
 * Utility factory class to create instances of common complex objects required by Download Actions.
 */
@Data
@Builder
public class DownloadWorkflowModule  {

  public static final String CONF_FILE = "download.properties";

  //Prefix for static settings
  public static final String PROPERTIES_PREFIX = "occurrence.download.";

  private static final String ES_PREFIX = "es.";

  private static final String INDEX_LOCKING_PATH = "/indices/";

  private final WorkflowConfiguration workflowConfiguration;

  private final DownloadJobConfiguration downloadJobConfiguration;

  /**
   * DownloadPrepare action factory method.
   * This is the initial action that counts records and its output is used to decide if a download is processed through Hive or Es.
   */
  public DownloadPrepareAction downloadPrepareAction(DwcTerm dwcTerm, String wfPath) {
    return DownloadPrepareAction.builder().esClient(esClient())
            .esIndex(workflowConfiguration.getSetting(DefaultSettings.ES_INDEX_KEY))
            .smallDownloadLimit(workflowConfiguration.getIntSetting(DefaultSettings.MAX_RECORDS_KEY))
            .workflowConfiguration(workflowConfiguration)
            .occurrenceDownloadService(clientBuilder().build(OccurrenceDownloadClient.class))
            .esFieldMapper(EsFieldMapper.builder().nestedIndex(workflowConfiguration.isEsNestedIndex())
                             .searchType(workflowConfiguration.getEsIndexType()).build())
            .coreTerm(dwcTerm)
            .wfPath(wfPath)
            .build();
  }

  /**
   * GBIF Ws client factory.
   */
  public ClientBuilder clientBuilder() {<commit_msg>building ws client according to the download type<commit_after>/*
 * Licensed 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.gbif.occurrence.download.inject;

import org.gbif.api.model.event.Event;
import org.gbif.api.model.occurrence.DownloadFormat;
import org.gbif.api.model.occurrence.Occurrence;
import org.gbif.api.service.registry.OccurrenceDownloadService;
import org.gbif.dwc.terms.DwcTerm;
import org.gbif.event.search.es.SearchHitEventConverter;
import org.gbif.occurrence.download.conf.WorkflowConfiguration;
import org.gbif.occurrence.download.file.DownloadAggregator;
import org.gbif.occurrence.download.file.DownloadJobConfiguration;
import org.gbif.occurrence.download.file.DownloadMaster;
import org.gbif.occurrence.download.file.OccurrenceMapReader;
import org.gbif.occurrence.download.file.dwca.DwcaDownloadAggregator;
import org.gbif.occurrence.download.file.simplecsv.SimpleCsvDownloadAggregator;
import org.gbif.occurrence.download.file.specieslist.SpeciesListDownloadAggregator;
import org.gbif.occurrence.download.oozie.DownloadPrepareAction;
import org.gbif.occurrence.search.es.EsConfig;
import org.gbif.occurrence.search.es.EsFieldMapper;
import org.gbif.occurrence.search.es.SearchHitConverter;
import org.gbif.occurrence.search.es.SearchHitOccurrenceConverter;
import org.gbif.registry.ws.client.EventDownloadClient;
import org.gbif.registry.ws.client.OccurrenceDownloadClient;
import org.gbif.wrangler.lock.Mutex;
import org.gbif.wrangler.lock.ReadWriteMutexFactory;
import org.gbif.wrangler.lock.zookeeper.ZookeeperSharedReadWriteMutex;
import org.gbif.ws.client.ClientBuilder;
import org.gbif.ws.json.JacksonJsonObjectMapperProvider;

import java.io.IOException;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.Map;
import java.util.function.Function;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.http.HttpHost;
import org.elasticsearch.client.NodeSelector;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.sniff.SniffOnFailureListener;
import org.elasticsearch.client.sniff.Sniffer;

import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import lombok.Builder;
import lombok.Data;
import lombok.experimental.UtilityClass;

;

/**
 * Utility factory class to create instances of common complex objects required by Download Actions.
 */
@Data
@Builder
public class DownloadWorkflowModule  {

  public static final String CONF_FILE = "download.properties";

  //Prefix for static settings
  public static final String PROPERTIES_PREFIX = "occurrence.download.";

  private static final String ES_PREFIX = "es.";

  private static final String INDEX_LOCKING_PATH = "/indices/";

  private final WorkflowConfiguration workflowConfiguration;

  private final DownloadJobConfiguration downloadJobConfiguration;

  /**
   * DownloadPrepare action factory method.
   * This is the initial action that counts records and its output is used to decide if a download is processed through Hive or Es.
   */
  public DownloadPrepareAction downloadPrepareAction(DwcTerm dwcTerm, String wfPath) {
    return DownloadPrepareAction.builder().esClient(esClient())
            .esIndex(workflowConfiguration.getSetting(DefaultSettings.ES_INDEX_KEY))
            .smallDownloadLimit(workflowConfiguration.getIntSetting(DefaultSettings.MAX_RECORDS_KEY))
            .workflowConfiguration(workflowConfiguration)
            .occurrenceDownloadService(downloadServiceClient(dwcTerm))
            .esFieldMapper(EsFieldMapper.builder().nestedIndex(workflowConfiguration.isEsNestedIndex())
                             .searchType(workflowConfiguration.getEsIndexType()).build())
            .coreTerm(dwcTerm)
            .wfPath(wfPath)
            .build();
  }

  /**
   * Creates a DownloadService for Event or Occurrence downloads.
   */
  private OccurrenceDownloadService downloadServiceClient(DwcTerm coreTerm) {
    return  DwcTerm.Event == coreTerm? clientBuilder().build(EventDownloadClient.class) : clientBuilder().build(OccurrenceDownloadClient.class);
  }

  /**
   * GBIF Ws client factory.
   */
  public ClientBuilder clientBuilder() {
{'fraction_non_alphanumeric': 0.08757333333333334, 'fraction_numerical': 0.0008533333333333333, 'mean_word_length': 5.041237113402062, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 75, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7595078', 'n_tokens_mistral': 2675, 'n_tokens_neox': 2633, 'n_words': 584}