add turbo_stream_from (#18)

docs/source/template-tags.md

@@ -87,3 +87,11 @@ return TurboStreamResponse(html)
 ```
 
 The code in Django view would be much cleaner and easier to maintain.
+
+## turbo_stream_from
+
+This can help render `turbo-cable-stream-source` in Django template
+
+`<turbo-cable-stream-source>` is a custom element provided by [turbo-rails](https://github.com/hotwired/turbo-rails/blob/097d8f90cf0c5ed24ac6b1a49cead73d49fa8ab5/app/javascript/turbo/cable_stream_source_element.js), with it, we can send Turbo Stream over the websocket connection and update the page in real time.
+
+The `<turbo-cable-stream-source>` is built on Rails ActionCable, which provide many great feature out of the box, such as `Automatic Reconnection`, so we can focus on the business logic.

src/turbo_response/channel_helper.py

@@ -0,0 +1,83 @@
+import json
+from typing import List, Tuple, Union
+
+from django.core import signing
+from django.core.signing import Signer
+from django.template.loader import render_to_string
+
+from .templatetags.turbo_helper import dom_id
+
+try:
+    from asgiref.sync import async_to_sync
+    from channels.layers import get_channel_layer
+except ImportError as err:
+    raise Exception("Please make sure django-channels is installed") from err
+
+
+signer = Signer()
+
+
+def stream_name_from(streamables: Union[List, object]) -> str:
+    """
+    Generate stream_name from a list of objects or a single object.
+    """
+    if isinstance(streamables, list):
+        return "_".join(stream_name_from(streamable) for streamable in streamables)
+    else:
+        return dom_id(streamables)
+
+
+def generate_signed_stream_key(stream_name: str) -> str:
+    """
+    Generate signed stream key from stream_name
+    """
+    return signer.sign(stream_name)
+
+
+def verify_signed_stream_key(signed_stream_key: str) -> Tuple[bool, str]:
+    """
+    Verify signed stream key
+    """
+    try:
+        unsigned_data = signer.unsign(signed_stream_key)
+        return True, unsigned_data
+
+    except signing.BadSignature:
+        pass
+
+    return False, ""
+
+
+def generate_channel_group_name(channel: str, stream_name: str):
+    """
+    Generate Django Channel group name from channel and stream_name
+    """
+    return f"{channel}_{stream_name}"
+
+
+def broadcast_render_to(streamables: Union[List, object], template: str, context=None):
+    if context is None:
+        context = {}
+
+    html = render_to_string(template, context=context)
+
+    stream_name = stream_name_from(streamables)
+    channel_group_name = generate_channel_group_name("TurboStreamsChannel", stream_name)
+
+    channel_layer = get_channel_layer()
+    async_to_sync(channel_layer.group_send)(
+        channel_group_name,
+        {
+            "type": "turbo_stream_message",
+            "data": {
+                "identifier": json.dumps(
+                    {
+                        "channel": "TurboStreamsChannel",
+                        "signed_stream_name": generate_signed_stream_key(stream_name),
+                    },
+                    separators=(",", ":"),
+                ),
+                "message": html,
+            },
+        },
+    )

src/turbo_response/templatetags/turbo_helper.py

@@ -108,6 +108,29 @@ def render(self, context):
         return django_engine.from_string(template_string).render(context)
 
 
+class TurboStreamFromTagNode(Node):
+    def __init__(self, stream_name_array):
+        self.stream_name_array = stream_name_array
+
+    def __repr__(self):
+        return "<%s>" % self.__class__.__name__
+
+    def render(self, context):
+        from ..channel_helper import generate_signed_stream_key, stream_name_from
+
+        stream_name_array = [
+            stream_name.resolve(context) for stream_name in self.stream_name_array
+        ]
+        stream_name_string = stream_name_from(stream_name_array)
+
+        django_engine = engines["django"]
+        template_string = """<turbo-cable-stream-source channel="TurboStreamsChannel" signed-stream-name="{{ signed_stream_name }}"></turbo-cable-stream-source>"""
+        context = {
+            "signed_stream_name": generate_signed_stream_key(stream_name_string),
+        }
+        return django_engine.from_string(template_string).render(context)
+
+
 @register.tag("turbo_frame")
 def turbo_frame_tag(parser, token):
     args = token.split_contents()
@@ -173,3 +196,18 @@ def turbo_stream_tag(parser, token):
     parser.delete_first_token()
 
     return TurboStreamTagNode(action, target, nodelist, extra_context=extra_context)
+
+
+@register.tag("turbo_stream_from")
+def turbo_stream_from_tag(parser, token):
+    args = token.split_contents()
+
+    if len(args) < 1:
+        raise TemplateSyntaxError(
+            "'turbo_stream_from' tag requires at least one arguments"
+        )
+
+    remaining_bits = args[1:]
+    stream_name_array = [parser.compile_filter(bit) for bit in remaining_bits]
+
+    return TurboStreamFromTagNode(stream_name_array)

tests/conftest.py

@@ -31,10 +31,10 @@ def pytest_configure():
             "django.contrib.sessions",
             "django.contrib.sites",
             "turbo_response",
+            "channels",
             "tests.testapp.apps.TestAppConfig",
         ],
         ROOT_URLCONF="tests.testapp.urls",
-        TURBO_RESPONSE_RENDERER="turbo_response.renderers.TemplatesSetting",
     )
 
 

tests/test_tags.py

@@ -101,3 +101,29 @@ def test_dom_id_variable(self):
             output
             == '<turbo-stream action="append" target="test"><template>Test</template></turbo-stream>'
         )
+
+
+class TestStreamFrom:
+    def test_string(self):
+        template = """
+        {% load turbo_helper %}
+
+        {% turbo_stream_from "test" %}
+        """
+        output = render(template, {}).strip()
+        assert (
+            output
+            == '<turbo-cable-stream-source channel="TurboStreamsChannel" signed-stream-name="test:1DyYXz2Y_VbgIPXC1AQ0ZGHhAx71uaZ36r4DFwXDaiU"></turbo-cable-stream-source>'
+        )
+
+    def test_dom_id_variable(self):
+        template = """
+        {% load turbo_helper %}
+
+        {% turbo_stream_from "test" dom_id %}
+        """
+        output = render(template, {"dom_id": "todo_3"}).strip()
+        assert (
+            output
+            == '<turbo-cable-stream-source channel="TurboStreamsChannel" signed-stream-name="test_todo_3:7ZS0MxQWhRTCAnG3olGO9AJKfvos3iaHGoBMBt8ZbSM"></turbo-cable-stream-source>'
+        )

tox.ini

@@ -8,6 +8,7 @@ changedir=tests
 deps =
     django32: django>=3.2,<3.3
     django42: django>=3.3,<4.3
+    channels
     typing_extensions
     pytest
     pytest-django