using Elasticsearch.Net;

using log4net;

using Newtonsoft.Json.Linq;

using System;

using System.Collections.Generic;

using System.IO;

using System.Linq;

using System.Text;

public class Program

{

        private readonly static ILog log = null;

        public string Name { get; set; }

        public enum ConnectionPoolTypes { Static, Sniffying, Sticky }

        public ConnectionPoolTypes ConnectionPoolType { get; set; }

​

        public string ConnectionString { get; set; }

        public int RequestTimeOut { get; set; }

​

        public string UserName { get; set; }

        public string Password { get; set; }

        public string LogPath { get; set; }

    public static void Main()

    {

        Console.WriteLine("Hello World");

    }

    protected ElasticLowLevelClient esClient;

​

    public void Connect()

        {

            var uris = this.ConnectionString.Split(',').Select(uri => new Uri(uri));

​

            IConnectionPool pool = null;

            if (this.ConnectionPoolType == ElasticSearchBase.ConnectionPoolTypes.Sniffying)

            {

                pool = new SniffingConnectionPool(uris);

            }

            else if (this.ConnectionPoolType == ElasticSearchBase.ConnectionPoolTypes.Sticky)

            {

                pool = new StickyConnectionPool(uris);

            }

            else

            {

                pool = new StaticConnectionPool(uris);

            }

​

            var conn = new ConnectionConfiguration(pool);

​

            conn.RequestTimeout(TimeSpan.FromSeconds(System.Convert.ToInt32(this.RequestTimeOut)))

                .BasicAuthentication(this.UserName, this.Password)

                .DisablePing()

                .ThrowExceptions();

​

            log.Info($"{this.Name} - Initialize ES client");

            this.esClient = new ElasticLowLevelClient(conn);

        }

    public JObject ScrollSearch(string indexName, string query, int scrollSpan)

        {

            string savedScrollId = null;   // scroll id saved for cleanup at the end

            try

            {

                if (this.esClient == null)

                {

                    throw new BnsException("No connection to ElasticSearch.");

                }

​

                log.Info($"{this.Name} - Scroll search: {query}");

                var searchResponse = this.esClient.Search<DynamicResponse>(indexName, query,

                    new SearchRequestParameters { Scroll = TimeSpan.FromSeconds(scrollSpan) });

​

                //looping/scrolling into the ES index for more records

                JArray resultArray = new JArray();

                while (searchResponse.Success && ((List<dynamic>)searchResponse.Body["hits"]["hits"]).Any())

                {

                    var stringResponse = Encoding.UTF8.GetString(searchResponse.ResponseBodyInBytes);

                    LogResponse(indexName, stringResponse);

                    var resp = JObject.Parse(stringResponse);

                    resultArray.Merge((JArray)((JObject)resp?["hits"])?["hits"]);

​

                    savedScrollId = searchResponse.Body["_scroll_id"].ToString();

                    log.Info($"{this.Name} - Scrolling");

                    searchResponse = esClient.Scroll<DynamicResponse>(PostData.Serializable(

                        new

                        {

                            scroll = $"{scrollSpan / 60}m",

                            scroll_id = savedScrollId

                        }));

                }

​

                if (!searchResponse.Success)

                {

                    throw searchResponse.OriginalException;

                }

​

                log.Info($"{this.Name} - Total records: {resultArray.Count}");

​

                // compose a JSON with the array content in propety 'root'

                var json = new JObject(new JProperty("root", resultArray));

                LogMessage(indexName, json);

                return json;

            }

            catch (Exception ex)

            {

                log.Error(ex);

                throw;

            }

            finally

            {

                log.Info($"{this.Name} - To clear the scroll once finished");

​

                if (savedScrollId != null)

                {

                    this.esClient?.ClearScroll<DynamicResponse>(PostData.Serializable(new { scroll_id = savedScrollId }));

                }

            }

​

        }

​

        public JObject NonScrollSearch(string indexName, string query)

        {

            try

            {

                if (this.esClient == null)

                {

                    throw new BnsException("No connection to ElasticSearch.");

                }

​

                log.Info($"{this.Name} - Non-scroll search index: {indexName} query: {query}");

                var searchResponse = this.esClient.Search<DynamicResponse>(indexName, query);

                if (!searchResponse.Success)

                {

                    throw searchResponse.OriginalException;

                }

                var stringResponse = Encoding.UTF8.GetString(searchResponse.ResponseBodyInBytes);

                LogResponse(indexName, stringResponse);

​

                JArray resultArray = new JArray();

                var resp = JObject.Parse(stringResponse);

                resultArray.Merge((JArray)((JObject)resp?["hits"])?["hits"]);

                log.Info($"{this.Name} - Total records: {resultArray.Count}");

​

                // compose a JSON with the array content in propety 'root'

                var json = new JObject(new JProperty("root", resultArray));

                LogMessage(indexName, json);

                return json;

            }

            catch (Exception ex)

            {

                log.Error(ex);

                throw;

            }

        }

     public string GetFieldMapping(string indexName, string field)

        {

            try

            {

                if (this.esClient == null)

                {

                    throw new BnsException("No connection to ElasticSearch.");

                }

​

                log.Info($"{this.Name} - GetFieldMapping for index: {indexName}, field: {field}");

                var response = this.esClient.Indices.GetFieldMapping<StringResponse>(indexName, field);

                if (!response.Success)

                {

                    log.Error($"{this.Name} - Error when GetFieldMapping index: {indexName}, field: {field}");

                    throw response.OriginalException;

                }

                LogResponse(indexName, response.Body);

​

                var resp = JObject.Parse(Encoding.UTF8.GetString(response.ResponseBodyInBytes));

​

                // when thismethod is called using an alias name that is mapped to multiple indices,

                // the response will contain mulitple mapping records for each index.

                // assuming all indexes have to same type, so only returning the type of the first index.

                return resp.SelectTokens("$..type")?.First()?.ToString();

            }

            catch (Exception ex)

            {

                log.Error(ex);

                throw;

            }

        }

​

        public void XmlDeserialize(XmlSerializationContext context)

        {

            var connCtx = context.GetChildContext("connection", true);

​

            var poolType = connCtx.ReadElementString("connectionPoolType", "Sticky");

            if ("Sniffing".Equals(poolType))

            {

                this.ConnectionPoolType = ConnectionPoolTypes.Sniffying;

            }

            else if ("Sticky".Equals(poolType))

            {

                this.ConnectionPoolType = ConnectionPoolTypes.Sticky;

            }

            else

            {

                this.ConnectionPoolType = ConnectionPoolTypes.Static;  // default

            }

​

            this.ConnectionString = connCtx.ReadElementString("connectionString");

​

            this.RequestTimeOut = connCtx.ReadElementAsInt32("requestTimeOut", 120); //default value: 2 mins

​

            this.UserName = connCtx.ReadElementString("userName");

​

            var passwordFileLocation = connCtx.ReadElementString("passwordLocation");

            string tempPass = File.ReadAllText(passwordFileLocation).Trim();

​

            bool encrypt = connCtx.GetChildContext("passwordLocation").ReadAttributeAsBool("encrypted", false);

            this.Password = encrypt ? BNSUtility.Decrypt(tempPass) : tempPass;

​

            this.LogPath = connCtx.ReadElementString("logPath", String.Empty);

        }

​

        public void XmlSerialize(XmlSerializationContext context)

        {

            throw new NotImplementedException();

        }

​

        private void LogResponse(string indexName, string repsonse)

        {

            // Log the first 1k and last 1k characters if response is longer than 2k characters

            var logString = repsonse?.Length >= 2048 ? (repsonse.Substring(0, 1024) + "\n---OMITTED---\n" + repsonse.Substring(repsonse.Length - 1024)) : repsonse;

            log.Info($"{this.Name} - Index: {indexName} Response: {repsonse?.Length} characters, content:\n{logString}");

        }

​

        private void LogMessage(string indexName, JObject json)

        {

            try

            {

                if (String.IsNullOrEmpty(this.LogPath))

                    return;

​

                Dictionary<string, object> parameters = new Dictionary<string, object>

                {

                    { "Now", DateTime.Now },

                    { "IndexName", indexName }

                };

                var path = FileUtility.GetPath(this.LogPath, parameters);

​

                if (!Directory.Exists(Path.GetDirectoryName(path)))

                    Directory.CreateDirectory(Path.GetDirectoryName(path));

​

                File.WriteAllText(path, json.ToString());

            }

            catch (Exception ex)

            {

                log.Error(ex);

            }

        }

}