**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
airflow/providers/amazon/aws/sensors/sqs.py
"""
        Check for message on subscribed queue and write to xcom the message with key ``messages``
        :param context: the context object
        :type context: dict
        :return: ``True`` if message is available or ``False``
        """
        sqs_conn = self.get_hook().get_conn()
        self.log.info('SQSSensor checking for message on queue: %s', self.sqs_queue)
        messages = sqs_conn.receive_message(
            QueueUrl=self.sqs_queue,
            MaxNumberOfMessages=self.max_messages,
            WaitTimeSeconds=self.wait_time_seconds,
        )
        self.log.info("received message %s", str(messages))
        if 'Messages' in messages and messages['Messages']:
            entries = [
                {'Id': message['MessageId'], 'ReceiptHandle': message['ReceiptHandle']}
                for message in messages['Messages']
            ]
            result = sqs_conn.delete_message_batch(QueueUrl=self.sqs_queue, Entries=entries)
            if 'Successful' in result:
                context['ti'].xcom_push(key='messages', value=messages)
                return True
            else:
                raise AirflowException(
                    'Delete SQS Messages failed ' + str(result) + ' for messages ' + str(messages)
                )
        return False
    def get_hook(self) -> SQSHook: