/// <summary>
        /// Constructs the transport with the specified settings
        /// </summary>
        public AmazonSqsTransport(string inputQueueAddress, string accessKeyId, string secretAccessKey, AmazonSQSConfig amazonSqsConfig, IRebusLoggerFactory rebusLoggerFactory, IAsyncTaskFactory asyncTaskFactory)
        {
            if (accessKeyId == null) throw new ArgumentNullException(nameof(accessKeyId));
            if (secretAccessKey == null) throw new ArgumentNullException(nameof(secretAccessKey));
            if (amazonSqsConfig == null) throw new ArgumentNullException(nameof(amazonSqsConfig));
            if (rebusLoggerFactory == null) throw new ArgumentNullException(nameof(rebusLoggerFactory));

            Address = inputQueueAddress;

            _log = rebusLoggerFactory.GetCurrentClassLogger();

            if (Address != null)
            {
                if (Address.Contains("/") && !Uri.IsWellFormedUriString(Address, UriKind.Absolute))
                {
                    throw new ArgumentException(
                        "You could either have a simple queue name without slash (eg. \"inputqueue\") - or a complete URL for the queue endpoint. (eg. \"https://sqs.eu-central-1.amazonaws.com/234234234234234/somqueue\")",
                        nameof(inputQueueAddress));
                }
            }

            _accessKeyId = accessKeyId;
            _secretAccessKey = secretAccessKey;
            _amazonSqsConfig = amazonSqsConfig;
            _asyncTaskFactory = asyncTaskFactory;
        }

        public bool Execute()
        {
            Console.WriteLine($"Moving {Count} messages from {SourceQueueName} to {DestinationQueueName} in {Region}.");

            var config = new AmazonSQSConfig { RegionEndpoint = RegionEndpoint.GetBySystemName(Region) };
            var client = new DefaultAwsClientFactory().GetSqsClient(config.RegionEndpoint);
            var sourceQueue = new SqsQueueByName(config.RegionEndpoint, SourceQueueName, client, JustSayingConstants.DEFAULT_HANDLER_RETRY_COUNT);
            var destinationQueue = new SqsQueueByName(config.RegionEndpoint, DestinationQueueName, client, JustSayingConstants.DEFAULT_HANDLER_RETRY_COUNT);

            EnsureQueueExists(sourceQueue);
            EnsureQueueExists(destinationQueue);

            var messages = PopMessagesFromSourceQueue(sourceQueue);
            var receiptHandles = messages.ToDictionary(m => m.MessageId, m => m.ReceiptHandle);
            
            var sendResponse = destinationQueue.Client.SendMessageBatch(new SendMessageBatchRequest
            {
                QueueUrl = destinationQueue.Url,
                Entries = messages.Select(x => new SendMessageBatchRequestEntry { Id = x.MessageId, MessageBody = x.Body }).ToList()
            });

            var deleteResponse = sourceQueue.Client.DeleteMessageBatch(new DeleteMessageBatchRequest
            {
                QueueUrl = sourceQueue.Url,
                Entries = sendResponse.Successful.Select(x => new DeleteMessageBatchRequestEntry
                {
                    Id = x.Id,
                    ReceiptHandle = receiptHandles[x.Id]
                }).ToList()
            });

            Console.WriteLine($"Moved {sendResponse.Successful.Count} messages from {SourceQueueName} to {DestinationQueueName} in {Region}.");

            return true;
        }

 public ResponseSender()
 {
     // initialize Amazon SQSClient
     AmazonSQSConfig sqsConfig = new AmazonSQSConfig();
     sqsConfig.ServiceURL = ConfigurationManager.AppSettings["SQSServiceURL"].ToString();
     m_sqsClient = AWSClientFactory.CreateAmazonSQSClient(sqsConfig);
 }

        public AmazonSQS(string keyId, string secretKey)
        {
            sqsConfig = new AmazonSQSConfig();
            //sqsConfig.ServiceURL = "https://sqs.amazonaws.com";

            client = AWSClientFactory.CreateAmazonSQSClient(keyId, secretKey, sqsConfig);
            queueUrls = new Dictionary<string, string>();
        }

        public string CreateQueue(string queueName)
        {
            var amazonSqsConfig = new AmazonSQSConfig {ServiceURL = ServiceUrl};

            using (var sqsClient = _awsConfig.CreateAwsClient<AmazonSQSClient>(amazonSqsConfig))
            {
                var response = sqsClient.CreateQueue(new CreateQueueRequest(queueName));

                return response.QueueUrl;
            }
        }

        /// <summary>
        /// Configures Rebus to use Amazon Simple Queue Service as the message transport
        /// </summary>
        public static void UseAmazonSqsAsOneWayClient(this StandardConfigurer<ITransport> configurer, string accessKeyId, string secretAccessKey, AmazonSQSConfig amazonSqsConfig)
        {
            configurer.Register(c =>
            {
                var rebusLoggerFactory = c.Get<IRebusLoggerFactory>();
                var asyncTaskFactory = c.Get<IAsyncTaskFactory>();

                return new AmazonSqsTransport(null, accessKeyId, secretAccessKey, amazonSqsConfig, rebusLoggerFactory, asyncTaskFactory);
            });

            OneWayClientBackdoor.ConfigureOneWayClient(configurer);
        }

 /// <summary>
 /// Instantiates the Poller.
 /// </summary>
 /// <param name="props">
 /// A <see cref="MessageGearsProperties"/>
 /// </param>
 /// <param name="listener">
 /// A <see cref="MessageGearsListener"/>
 /// </param>
 /// <param name="myAwsAccountKey">
 /// You AWS Account Key
 /// </param>
 /// <param name="myAwsSecretKey">
 /// Your AWS Secret Key
 /// </param>
 public MessageGearsAwsQueuePoller(MessageGearsAwsProperties props, MessageGearsListener listener)
 {
     this.props = props;
     this.emptyQueueDelayMillis = props.EmptyQueuePollingDelaySecs * 1000;
     this.listener = listener;
     AmazonSQSConfig config = new AmazonSQSConfig().WithMaxErrorRetry(props.SQSMaxErrorRetry);
     this.sqs = AWSClientFactory.CreateAmazonSQSClient (props.MyAWSAccountKey, props.MyAWSSecretKey, config);
     this.receiveMessageRequest = new ReceiveMessageRequest ()
         .WithQueueUrl (props.MyAWSEventQueueUrl)
         .WithMaxNumberOfMessages (props.SQSMaxBatchSize)
         .WithAttributeName("ApproximateReceiveCount")
         .WithVisibilityTimeout(props.SQSVisibilityTimeoutSecs);
     this.deleteMessageRequest = new DeleteMessageRequest().WithQueueUrl(props.MyAWSEventQueueUrl);
 }

        static AmazonSqsTransport CreateTransport(string inputQueueAddress, TimeSpan peeklockDuration)
        {
            var amazonSqsConfig = new AmazonSQSConfig
            {
                RegionEndpoint = RegionEndpoint.GetBySystemName(ConnectionInfo.RegionEndpoint)
            };

            var consoleLoggerFactory = new ConsoleLoggerFactory(false);
            var transport = new AmazonSqsTransport(inputQueueAddress, ConnectionInfo.AccessKeyId, ConnectionInfo.SecretAccessKey,
                amazonSqsConfig,
                consoleLoggerFactory,
                new TplAsyncTaskFactory(consoleLoggerFactory));

            transport.Initialize(peeklockDuration);
            transport.Purge();
            return transport;
        }

        protected override void SetUp()
        {
            var connectionInfo = AmazonSqsTransportFactory.ConnectionInfo;

            var accessKeyId = connectionInfo.AccessKeyId;
            var secretAccessKey = connectionInfo.SecretAccessKey;
            var amazonSqsConfig = new AmazonSQSConfig
            {
                RegionEndpoint = RegionEndpoint.GetBySystemName(AmazonSqsTransportFactory.ConnectionInfo.RegionEndpoint)
            };

            _activator = Using(new BuiltinHandlerActivator());

            Configure.With(_activator)
                .Transport(t => t.UseAmazonSqs(accessKeyId, secretAccessKey, amazonSqsConfig, TestConfig.QueueName("defertest")))
                .Options(o => o.LogPipeline())
                .Start();
        }

		/// <summary>
		/// Constructor taking the landing zone
		/// </summary>
		public DataRouterReportQueue(string InQueueName, string InLandingZoneTempPath, int InDecimateWaitingCountStart, int InDecimateWaitingCountEnd)
			: base(InQueueName, InLandingZoneTempPath, InDecimateWaitingCountStart, InDecimateWaitingCountEnd)
		{
			AWSCredentials Credentials = new StoredProfileAWSCredentials(Config.Default.AWSProfileName, Config.Default.AWSCredentialsFilepath);

			AmazonSQSConfig SqsConfig = new AmazonSQSConfig
			{
				ServiceURL = Config.Default.AWSSQSServiceURL
			};

			SqsClient = new AmazonSQSClient(Credentials, SqsConfig);

			AmazonS3Config S3Config = new AmazonS3Config
			{
				ServiceURL = Config.Default.AWSS3ServiceURL
			};

			S3Client = new AmazonS3Client(Credentials, S3Config);
		}

		public SqsActor ()
		{
			Receive<string> (x => {


                var sqs_url = Environment.GetEnvironmentVariable("sqs_url", EnvironmentVariableTarget.Process);
                var config = new AmazonSQSConfig();
                config.ServiceURL = sqs_url;

                var creds = new StoredProfileAWSCredentials();
                var client = new AmazonSQSClient(creds, config);

                var msg =  x + " and what " + Guid.NewGuid().ToString();
                var queue_url = Environment.GetEnvironmentVariable("queue_url", EnvironmentVariableTarget.Process);
             
                var request = new Amazon.SQS.Model.SendMessageRequest(queue_url, msg);
             
                client.SendMessage(request);

				Sender.Tell(string.Format("done  : [{0}]", msg ));
			});
		}

        /// <summary>
        /// Configures Rebus to use Amazon Simple Queue Service as the message transport
        /// </summary>
        public static void UseAmazonSqs(this StandardConfigurer<ITransport> configurer, string accessKeyId, string secretAccessKey, AmazonSQSConfig config, string inputQueueAddress)
        {
            configurer.Register(c =>
            {
                var rebusLoggerFactory = c.Get<IRebusLoggerFactory>();
                var asyncTaskFactory = c.Get<IAsyncTaskFactory>();

                return new AmazonSqsTransport(inputQueueAddress, accessKeyId, secretAccessKey, config, rebusLoggerFactory, asyncTaskFactory);
            });

            configurer
                .OtherService<IPipeline>()
                .Decorate(p =>
                {
                    var pipeline = p.Get<IPipeline>();

                    return new PipelineStepRemover(pipeline)
                        .RemoveIncomingStep(s => s.GetType() == typeof (HandleDeferredMessagesStep));
                });

            configurer.OtherService<ITimeoutManager>().Register(c => new DisabledTimeoutManager(), description: SqsTimeoutManagerText);
        }

        public CraneChatRequestSender()
        {
            // initialize Amazon SQSClient
            AmazonSQSConfig sqsConfig = new AmazonSQSConfig();
            sqsConfig.ServiceURL = ConfigurationManager.AppSettings["SQSServiceURL"].ToString();
            m_sqsClient = AWSClientFactory.CreateAmazonSQSClient(sqsConfig);

            // create 'Request' queue and save its URL
            if (null != m_sqsClient)
            {
                try
                {
                    CreateQueueRequest createQueueRequest = new CreateQueueRequest().WithQueueName("Request");
                    CreateQueueResponse createQueueResponse = m_sqsClient.CreateQueue(createQueueRequest);
                    m_requestQueueUrl = createQueueResponse.CreateQueueResult.QueueUrl;
                }
                catch (AmazonSQSException /*sqsException*/)
                {
                    throw;
                }
            }
        }

        public ISubscribeReponseMessage Subscribe(string queueUrl)
        {
            var sqsConfig = new AmazonSQSConfig {ServiceURL = SqsServiceUrl};
            var snsConfig = new AmazonSimpleNotificationServiceConfig {ServiceURL = SnsServiceUrl};

            using (var sqsClient = _awsConfig.CreateAwsClient<AmazonSQSClient>(sqsConfig))
            {
                var attributesRequest = new GetQueueAttributesRequest(queueUrl, new List<string> {"QueueArn"});

                var attributesResponse = sqsClient.GetQueueAttributes(attributesRequest);

                using (var snsClient = _awsConfig.CreateAwsClient<AmazonSimpleNotificationServiceClient>(snsConfig))
                {
                    var subribeResonse =
                        snsClient.Subscribe(new SubscribeRequest(TopicArn, "sqs", attributesResponse.QueueARN));
                }

                var actions = new ActionIdentifier[2];
                actions[0] = SQSActionIdentifiers.SendMessage;
                actions[1] = SQSActionIdentifiers.ReceiveMessage;
                var sqsPolicy =
                    new Policy().WithStatements(
                        new Statement(Statement.StatementEffect.Allow).WithPrincipals(Principal.AllUsers)
                            .WithResources(
                                new Resource(attributesResponse.QueueARN))
                            .WithConditions(
                                ConditionFactory.NewSourceArnCondition(
                                    TopicArn))
                            .WithActionIdentifiers(actions));
                var setQueueAttributesRequest = new SetQueueAttributesRequest();

                var attributes = new Dictionary<string, string> {{"Policy", sqsPolicy.ToJson()}};
                var attRequest = new SetQueueAttributesRequest(attributesRequest.QueueUrl, attributes);

                sqsClient.SetQueueAttributes(attRequest);

                return new SubcriptionMessage("Ok");
            }
        }

 /// <summary>
 /// Constructs AmazonSQSClient with AWS Credentials and an
 /// AmazonSQSClient Configuration object.
 /// </summary>
 /// <param name="credentials">AWS Credentials</param>
 /// <param name="clientConfig">The AmazonSQSClient Configuration Object</param>
 public AmazonSQSClient(AWSCredentials credentials, AmazonSQSConfig clientConfig)
     : base(credentials, clientConfig, AuthenticationTypes.User | AuthenticationTypes.Session)
 {
 }

 /// <summary>
 /// Create a client for the Amazon SQS Service with AWSCredentials and an AmazonSQS Configuration object.
 /// </summary>
 /// <param name="credentials">AWS Credentials</param>
 /// <param name="config">Configuration options for the service like HTTP Proxy, # of connections, etc</param>
 /// <returns>An Amazon SQS client</returns>
 /// <remarks>
 /// </remarks>
 public static IAmazonSQS CreateAmazonSQSClient(AWSCredentials credentials, AmazonSQSConfig config)
 {
     return new AmazonSQSClient(credentials, config);
 }

 /// <summary>
 /// Create a client for the Amazon SQS Service with the credentials loaded from the application's
 /// default configuration, and if unsuccessful from the Instance Profile service on an EC2 instance.
 /// 
 /// Example App.config with credentials set. 
 /// <code>
 /// &lt;?xml version="1.0" encoding="utf-8" ?&gt;
 /// &lt;configuration&gt;
 ///     &lt;appSettings&gt;
 ///         &lt;add key="AWSAccessKey" value="********************"/&gt;
 ///         &lt;add key="AWSSecretKey" value="****************************************"/&gt;
 ///     &lt;/appSettings&gt;
 /// &lt;/configuration&gt;
 /// </code>
 /// </summary>
 /// <param name="config">Configuration options for the service like HTTP Proxy, # of connections, etc</param>
 /// <returns>An Amazon SQS client</returns>
 public static AmazonSQS CreateAmazonSQSClient(AmazonSQSConfig config)
 {
     return new AmazonSQSClient(config);
 }

		public SqsController()
		{
			var config = new AmazonSQSConfig();
			config.ServiceURL = "https://sqs.eu-central-1.amazonaws.com/";			
			_client = new AmazonSQSClient(ConfigurationManager.AppSettings["AWSAccessKey"], ConfigurationManager.AppSettings["AWSSecretAccessKey"], config);
		}

 /// <summary>
 /// Constructs AmazonSQSClient with the credentials loaded from the application's
 /// default configuration, and if unsuccessful from the Instance Profile service on an EC2 instance.
 /// 
 /// Example App.config with credentials set. 
 /// <code>
 /// &lt;?xml version="1.0" encoding="utf-8" ?&gt;
 /// &lt;configuration&gt;
 ///     &lt;appSettings&gt;
 ///         &lt;add key="AWSAccessKey" value="********************"/&gt;
 ///         &lt;add key="AWSSecretKey" value="****************************************"/&gt;
 ///     &lt;/appSettings&gt;
 /// &lt;/configuration&gt;
 /// </code>
 ///
 /// </summary>
 /// <param name="config">The AmazonSQS Configuration Object</param>
 public AmazonSQSClient(AmazonSQSConfig config)
     : base(FallbackCredentialsFactory.GetCredentials(), config, AuthenticationTypes.User | AuthenticationTypes.Session) { }

 /// <summary>
 /// Constructs AmazonSQSClient with AWS Access Key ID, AWS Secret Key and an
 /// AmazonSQSClient Configuration object. 
 /// </summary>
 /// <param name="awsAccessKeyId">AWS Access Key ID</param>
 /// <param name="awsSecretAccessKey">AWS Secret Access Key</param>
 /// <param name="awsSessionToken">AWS Session Token</param>
 /// <param name="clientConfig">The AmazonSQSClient Configuration Object</param>
 public AmazonSQSClient(string awsAccessKeyId, string awsSecretAccessKey, string awsSessionToken, AmazonSQSConfig clientConfig)
     : base(awsAccessKeyId, awsSecretAccessKey, awsSessionToken, clientConfig)
 {
 }