Skip to content

EventHandler

hololinked.server.http.handlers.EventHandler

Bases: BaseHandler

handles events emitted by Thing and tunnels them as HTTP SSE.

Source code in repo/hololinked/hololinked/server/http/handlers.py 
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
class EventHandler(BaseHandler):
    """handles events emitted by `Thing` and tunnels them as HTTP SSE."""

    resource: EventAffordance | PropertyAffordance
    """an event, or the observable property whose change event is streamed - both name an event to subscribe to"""

    def initialize(
        self,
        resource: InteractionAffordance | EventAffordance,
        config: RuntimeConfig,
        logger: structlog.stdlib.BoundLogger,
        thing: Thing,
        metadata: HandlerMetadata | None = None,
    ) -> None:
        """
        Set up the handler to stream events with a plain SSE data header.

        Raises
        ------
        RuntimeError
            If the `Thing` is not served by an event loop.
        """
        super().initialize(resource, config, logger, thing, metadata)
        self.data_header = b"data: %s\n\n"
        if self.thing.eventloop is None:  # type gaurd, not a real logic
            raise RuntimeError("Thing is not served by an event loop")

    def set_custom_default_headers(self) -> None:
        """
        Sets default headers for event handling. The general headers are listed as follows:

        ```yaml
        Content-Type: text/event-stream
        Cache-Control: no-cache
        Connection: keep-alive
        Access-Control-Allow-Credentials: true # Possibly for cookie auth
        Access-Control-Allow-Origin: <client> # if CORS is enabled
        ```
        """  # noqa: D400
        self.set_header("Content-Type", "text/event-stream")
        self.set_header("Cache-Control", "no-cache")
        self.set_header("Connection", "keep-alive")
        super().set_custom_default_headers()

    async def get(self):
        """Events are support only with GET method."""
        if await self.has_access_control():
            self.set_custom_default_headers()
            await self.handle_datastream()
        self.finish()

    async def options(self):  # ty: ignore[invalid-method-override]
        """Options for the resource."""
        if await self.has_access_control():
            self.set_status(204)
            self.set_custom_default_headers()
            self.set_access_control_allow_headers()
            self.set_header("Access-Control-Allow-Methods", "GET")
        self.finish()

    async def handle_datastream(self) -> None:
        """Called by GET method and handles the event publishing."""  # noqa: DOC501
        if not self.thing.eventloop:  # type gaurd, not a real logic.
            raise RuntimeError("Thing is not served by an event loop")
        try:
            subscription = EventSubscription(
                self.thing.eventloop.event_bus,
                self.resource.event_unique_identifier,
            )
            self.set_status(200)
        except Exception as ex:
            self.logger.error(f"error while subscribing to event - {str(ex)}")
            self.set_status(500, f"could not subscribe to event source from thing - {str(ex)}")
            self.write(Serializers.json.dumps({"exception": format_exception_as_json(ex)}))
            return

        # Send the header right away. One needs to flush to even send headers.
        # This confirms that a subscription happened. If clients end with too short timeout even without receiving
        # headers, then they think that the subscription did not go through.
        await self.flush()

        try:
            while True:
                try:
                    data = await subscription.receive(timeout=10)
                    body, content_type = subscription.encode(data)
                    # TODO use content_type and get rid of other event handlers
                    self.write(self.data_header % body)
                    self.logger.debug(f"new data scheduled to flush - {self.resource.name}")
                    # flushes and handles heartbeat - raises StreamClosedError if the client left
                    await self.flush()
                except TimeoutError:
                    self.logger.debug(f"found no new data - {self.resource.name}")
                except StreamClosedError:
                    break
                except Exception as ex:
                    self.logger.error(f"error while pushing event - {str(ex)}")
                    self.write(self.data_header % Serializers.json.dumps({"exception": format_exception_as_json(ex)}))
        finally:
            subscription.unsubscribe()

Attributes

resource instance-attribute 

resource: EventAffordance | PropertyAffordance

an event, or the observable property whose change event is streamed - both name an event to subscribe to

Functions

initialize 

initialize(resource: InteractionAffordance | EventAffordance, config: RuntimeConfig, logger: BoundLogger, thing: Thing, metadata: HandlerMetadata | None = None) -> None

Set up the handler to stream events with a plain SSE data header.

Raises:

Type Description
RuntimeError

If the Thing is not served by an event loop.
Source code in repo/hololinked/hololinked/server/http/handlers.py 
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
def initialize(
    self,
    resource: InteractionAffordance | EventAffordance,
    config: RuntimeConfig,
    logger: structlog.stdlib.BoundLogger,
    thing: Thing,
    metadata: HandlerMetadata | None = None,
) -> None:
    """
    Set up the handler to stream events with a plain SSE data header.

    Raises
    ------
    RuntimeError
        If the `Thing` is not served by an event loop.
    """
    super().initialize(resource, config, logger, thing, metadata)
    self.data_header = b"data: %s\n\n"
    if self.thing.eventloop is None:  # type gaurd, not a real logic
        raise RuntimeError("Thing is not served by an event loop")

set_custom_default_headers 

set_custom_default_headers() -> None

Sets default headers for event handling. The general headers are listed as follows:

Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive
Access-Control-Allow-Credentials: true # Possibly for cookie auth
Access-Control-Allow-Origin: <client> # if CORS is enabled
Source code in repo/hololinked/hololinked/server/http/handlers.py 
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
def set_custom_default_headers(self) -> None:
    """
    Sets default headers for event handling. The general headers are listed as follows:

    ```yaml
    Content-Type: text/event-stream
    Cache-Control: no-cache
    Connection: keep-alive
    Access-Control-Allow-Credentials: true # Possibly for cookie auth
    Access-Control-Allow-Origin: <client> # if CORS is enabled
    ```
    """  # noqa: D400
    self.set_header("Content-Type", "text/event-stream")
    self.set_header("Cache-Control", "no-cache")
    self.set_header("Connection", "keep-alive")
    super().set_custom_default_headers()

get async 

get()

Events are support only with GET method.

Source code in repo/hololinked/hololinked/server/http/handlers.py 
744
745
746
747
748
749
async def get(self):
    """Events are support only with GET method."""
    if await self.has_access_control():
        self.set_custom_default_headers()
        await self.handle_datastream()
    self.finish()

options async 

options()

Options for the resource.

Source code in repo/hololinked/hololinked/server/http/handlers.py 
751
752
753
754
755
756
757
758
async def options(self):  # ty: ignore[invalid-method-override]
    """Options for the resource."""
    if await self.has_access_control():
        self.set_status(204)
        self.set_custom_default_headers()
        self.set_access_control_allow_headers()
        self.set_header("Access-Control-Allow-Methods", "GET")
    self.finish()

handle_datastream async 

handle_datastream() -> None

Called by GET method and handles the event publishing.

Source code in repo/hololinked/hololinked/server/http/handlers.py 
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
async def handle_datastream(self) -> None:
    """Called by GET method and handles the event publishing."""  # noqa: DOC501
    if not self.thing.eventloop:  # type gaurd, not a real logic.
        raise RuntimeError("Thing is not served by an event loop")
    try:
        subscription = EventSubscription(
            self.thing.eventloop.event_bus,
            self.resource.event_unique_identifier,
        )
        self.set_status(200)
    except Exception as ex:
        self.logger.error(f"error while subscribing to event - {str(ex)}")
        self.set_status(500, f"could not subscribe to event source from thing - {str(ex)}")
        self.write(Serializers.json.dumps({"exception": format_exception_as_json(ex)}))
        return

    # Send the header right away. One needs to flush to even send headers.
    # This confirms that a subscription happened. If clients end with too short timeout even without receiving
    # headers, then they think that the subscription did not go through.
    await self.flush()

    try:
        while True:
            try:
                data = await subscription.receive(timeout=10)
                body, content_type = subscription.encode(data)
                # TODO use content_type and get rid of other event handlers
                self.write(self.data_header % body)
                self.logger.debug(f"new data scheduled to flush - {self.resource.name}")
                # flushes and handles heartbeat - raises StreamClosedError if the client left
                await self.flush()
            except TimeoutError:
                self.logger.debug(f"found no new data - {self.resource.name}")
            except StreamClosedError:
                break
            except Exception as ex:
                self.logger.error(f"error while pushing event - {str(ex)}")
                self.write(self.data_header % Serializers.json.dumps({"exception": format_exception_as_json(ex)}))
    finally:
        subscription.unsubscribe()

JPEGImageEventHandler

Bases: EventHandler

handles events with images with JPEG image data header.

Source code in repo/hololinked/hololinked/server/http/handlers.py 
802
803
804
805
806
807
808
809
810
811
812
813
814
815
class JPEGImageEventHandler(EventHandler):
    """handles events with images with JPEG image data header."""

    def initialize(
        self,
        resource: InteractionAffordance | EventAffordance,
        config: RuntimeConfig,
        logger: structlog.stdlib.BoundLogger,
        thing: Thing,
        metadata: HandlerMetadata | None = None,
    ) -> None:
        """Set up the handler to stream events with a base64 JPEG image SSE data header."""
        super().initialize(resource, config, logger, thing, metadata)
        self.data_header = b"data:image/jpeg;base64,%s\n\n"

Functions

initialize 

initialize(resource: InteractionAffordance | EventAffordance, config: RuntimeConfig, logger: BoundLogger, thing: Thing, metadata: HandlerMetadata | None = None) -> None

Set up the handler to stream events with a base64 JPEG image SSE data header.

Source code in repo/hololinked/hololinked/server/http/handlers.py 
805
806
807
808
809
810
811
812
813
814
815
def initialize(
    self,
    resource: InteractionAffordance | EventAffordance,
    config: RuntimeConfig,
    logger: structlog.stdlib.BoundLogger,
    thing: Thing,
    metadata: HandlerMetadata | None = None,
) -> None:
    """Set up the handler to stream events with a base64 JPEG image SSE data header."""
    super().initialize(resource, config, logger, thing, metadata)
    self.data_header = b"data:image/jpeg;base64,%s\n\n"

PNGImageEventHandler

Bases: EventHandler

handles events with images with PNG image data header.

Source code in repo/hololinked/hololinked/server/http/handlers.py 
818
819
820
821
822
823
824
825
826
827
828
829
830
831
class PNGImageEventHandler(EventHandler):
    """handles events with images with PNG image data header."""

    def initialize(
        self,
        resource: InteractionAffordance | EventAffordance,
        config: RuntimeConfig,
        logger: structlog.stdlib.BoundLogger,
        thing: Thing,
        metadata: HandlerMetadata | None = None,
    ) -> None:
        """Set up the handler to stream events with a base64 PNG image SSE data header."""
        super().initialize(resource, config, logger, thing, metadata)
        self.data_header = b"data:image/png;base64,%s\n\n"

Functions

initialize 

initialize(resource: InteractionAffordance | EventAffordance, config: RuntimeConfig, logger: BoundLogger, thing: Thing, metadata: HandlerMetadata | None = None) -> None

Set up the handler to stream events with a base64 PNG image SSE data header.

Source code in repo/hololinked/hololinked/server/http/handlers.py 
821
822
823
824
825
826
827
828
829
830
831
def initialize(
    self,
    resource: InteractionAffordance | EventAffordance,
    config: RuntimeConfig,
    logger: structlog.stdlib.BoundLogger,
    thing: Thing,
    metadata: HandlerMetadata | None = None,
) -> None:
    """Set up the handler to stream events with a base64 PNG image SSE data header."""
    super().initialize(resource, config, logger, thing, metadata)
    self.data_header = b"data:image/png;base64,%s\n\n"