**Description**

Improve the AWS SQS Sensor as follows:
+ Add optional visibility_timeout parameter
+ Add a customisable / overrideable filter capability so we can filter/ignore irrelevant messages

[Not needed, see below conversation] 
--- Check the HTTP status code in AWS response and raise Exception if not 200 - best practice

**Use case / motivation**

I'd like to make the SQS sensor more flexible to enable the following use case:
+ A single queue can be used as a channel for messages from multiple event sources and or multiple targets
+ We need a way to filter and ignore messages not relevant to us, which other processes are looking for

**Are you willing to submit a PR?**

Yes, please assign to me

**Related Issues**

None
tests/providers/amazon/aws/sensors/test_sqs.py
def setUp(self):
        args = {'owner': 'airflow', 'start_date': DEFAULT_DATE}
        self.dag = DAG('test_dag_id', default_args=args)
        self.sensor = SQSSensor(
            task_id='test_task', dag=self.dag, sqs_queue='test', aws_conn_id='aws_default'
        )
        self.mock_context = mock.MagicMock()
        self.sqs_hook = SQSHook()
    @mock_sqs
    def test_poke_success(self):
        self.sqs_hook.create_queue('test')
        self.sqs_hook.send_message(queue_url='test', message_body='hello')
        result = self.sensor.poke(self.mock_context)
        assert result
        assert "'Body': 'hello'" in str(
            self.mock_context['ti'].method_calls
        ), "context call should contain message hello"
    @mock_sqs
    def test_poke_no_message_failed(self):
        self.sqs_hook.create_queue('test')
        result = self.sensor.poke(self.mock_context)
        assert not result
        context_calls = []
        assert self.mock_context['ti'].method_calls == context_calls, "context call  should be same"
    @mock.patch.object(SQSHook, 'get_conn')
    def test_poke_delete_raise_airflow_exception(self, mock_conn):