ColtenOuO commented on code in PR #72133: URL: https://github.com/apache/airflow/pull/72133#discussion_r3873878901
########## providers/standard/src/airflow/providers/standard/sensors/websocket.py: ########## @@ -0,0 +1,107 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you 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. +from __future__ import annotations + +import datetime +import time +from collections.abc import Sequence +from typing import TYPE_CHECKING, Any + +from websockets.sync.client import connect + +from airflow.providers.common.compat.sdk import BaseSensorOperator, conf, poke_mode_only +from airflow.providers.standard.triggers.websocket import WebSocketTrigger + +if TYPE_CHECKING: + from airflow.sdk import Context + + +@poke_mode_only +class WebSocketSensor(BaseSensorOperator): + """ + Waits for a message on a WebSocket connection. + + WebSocket messages are consumptive: once read, a message cannot be read again, and + reconnecting can duplicate ``message_to_send`` against the remote server. The + non-deferrable path therefore opens exactly one connection and blocks on it for up to + ``timeout`` seconds instead of reconnecting every ``poke_interval`` (which has no + effect on this sensor), and this sensor is marked poke-mode-only since a rescheduled + invocation would need a new connection anyway. + + :param url: The ``ws://`` or ``wss://`` URL of the WebSocket server to connect to. + :param header: Optional headers sent when opening the connection. + :param message_to_send: Optional message sent right after the connection is established. + :param deferrable: If waiting for completion, whether to defer the task until done, + default is ``False``. + + .. seealso:: + For more information on how to use this sensor, take a look at the guide: + :ref:`howto/operator:WebSocketSensor` Review Comment: reference need be changed -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
