We need to start documenting changes. This is not a Jupyter project, but it could be nice to use jupyter-releaser.
What do you think?

@dmonad could you give rights to the following people on this repo?

@afshin
@SylvainCorlay
@jtpio
@fcollonval

The YStore introduced in #18 should be independent of ypy-websocket. I think we should create a dedicated repository/package in the future.

I have a problem running tests and it seems that the problem is in this fixture:

The error is:

______________________________ test_ypy_yjs_1[1] _______________________________

yws_server = <async_generator object yws_server at 0x7f2dccd6db70>
yjs_client = <Popen: returncode: None args: ['node', 'tests/yjs_client_1.js']>

    @pytest.mark.asyncio
    @pytest.mark.parametrize("yjs_client", "1", indirect=True)
    async def test_ypy_yjs_1(yws_server, yjs_client):
        # wait for the JS client to connect
        tt, dt = 0, 0.1
        while True:
            await asyncio.sleep(dt)
>           if "/my-roomname" in yws_server.rooms:
E           AttributeError: 'async_generator' object has no attribute 'rooms'

From the error, it seems that the fixture produces a standard generator instead of the instance of the server.

I googled a little bit and it seems that this fixture should be marked with @pytest_asyncio.fixture instead of @pytest.fixture. Source: https://stackoverflow.com/questions/72996818/attributeerror-in-pytest-with-asyncio-after-include-code-in-fixtures

The publication of update is done task at a time:

We should do those update concurrently in a asyncio.gather to get them as quickly as possible and not be stuck by a slow client.

Moreover we should update with higher priority the document because it is the reference to be uploaded by any new clients before receiving deltas.

I think that part of the code is partly responsible for a data lost case seen when 20 people were collaborating simultaneously. What happen is some clients (cannot know if all) were still receiving the deltas. But the file on disk (that is a regular dump of the in-memory ydoc) stops updated. And if a new client connects, the document at the latest dump version was the one pushed.

My best guess (unfortunately we did not see any error) is that one client was blocking and so the code after the client loop was never executed.

cc: @davidbrochart @hbcarlos

Hello, I use WebsocketServer from ypy-websockets:

    while True:
        try:
            async with (
                WebsocketServer(log=logger) as websocket_server,
                serve(
                    my_ws_handler(websocket_server),
                    GeneralConfig.WS_SERVER_HOST,
                    5567,
                    ping_interval=None,
                ),
            ):
                await asyncio.Future()
        except Exception as exp:
            global PREV_COUNT_CLIENTS_WS
            print(websocket_server.rooms)
            PREV_COUNT_CLIENTS_WS = 0
            logger.error(str(exp))
            logger.info('Restart server')

When my client disconnects, the server crashes due to the below fault. I had to restart the server, what are the alternatives?
Error:
| websockets.exceptions.ConnectionClosedOK: received 1000 (OK); then sent 1000 (OK)

It seems to me that this package works fine with aiofiles 23 so the upper version limit can be higher or removed.

We must add a way to know if a store is compatible with the current version or not.

I would suggest writing a string in the header for the file dump and to use the pragma user_version for SQLite.

I am a software engineer at LinkedIn and my team is looking to leverage the functionalities of Real Time Collaboration (RTC) based on YJS within our custom Jupyter ecosystem.
Our use case requires RTC to work between multiple clients each of whom are on their own separate server (K8s pod).

To do this, we tried adding a MySQL implementation [closest to the SQLite implementation here] that stores the Y-Updates in a remote MySQL DB so that each of the servers use this same DB for content syncing.
The syncing does work but is highly unstable where sometimes the sync does not work or the cursor automatically moves to the beginning of the document or some other flaky edits are noticed.

Could this be due to MySQL not being the right choice for this use case?
We would appreciate any thoughts on this or suggestions on how to better choose a remote distributed YStore.

This would be more useful if the server were compatible with the ASGI spec. If it were, ypy-websocket could immediately be used in conjunction with popular web servers like Starlette, FastAPI, and Sanic.

Hi,

Thanks for all the great work on making CRDTs generally accessible to developers.

I tried the examples snippets for client and server in the README.md but the client fails with an exception.

server.py

import asyncio
from websockets import serve
from ypy_websocket import WebsocketServer

async def server():
    websocket_server = WebsocketServer()
    async with serve(websocket_server.serve, "localhost", 1234):
        await asyncio.Future()  # run forever

asyncio.run(server())

client.py

import asyncio
import y_py as Y
from websockets import connect
from ypy_websocket import WebsocketProvider

async def client():
    ydoc = Y.YDoc()
    async with connect("ws://localhost:1234/my-roomname") as websocket:
        WebsocketProvider(ydoc, websocket)
        ymap = ydoc.get_map("map")
        with ydoc.begin_transaction() as t:
            ymap.set(t, "key", "value")

asyncio.run(client())

fails with

Task exception was never retrieved
future: <Task finished coro=<WebsocketProvider._run() done, defined at /home/chris/.local/lib/python3.7/site-packages/ypy_websocket/websocket_provider.py:29> exception=ConnectionClosedOK(Close(code=1000, reason=''), Close(code=1000, reason=''), False)>
Traceback (most recent call last):
  File "/home/chris/.local/lib/python3.7/site-packages/ypy_websocket/websocket_provider.py", line 34, in _run
    await process_sync_message(message[1:], self._ydoc, self._websocket, self.log)
  File "/home/chris/.local/lib/python3.7/site-packages/ypy_websocket/yutils.py", line 120, in process_sync_message
    await websocket.send(reply)
  File "/home/chris/.local/lib/python3.7/site-packages/websockets/legacy/protocol.py", line 635, in send
    await self.ensure_open()
  File "/home/chris/.local/lib/python3.7/site-packages/websockets/legacy/protocol.py", line 953, in ensure_open
    raise self.connection_closed_exc()
websockets.exceptions.ConnectionClosedOK: sent 1000 (OK); then received 1000 (OK)

Any pointers why this could be?

Thanks,
Chris

Is it possible to update the YDoc on the server and broadcast changes to clients? The only way I can see to do this right now would be to connect from the server to itself over Websocket, which feels like a layer of indirection.

I've been able to do something like this, but once I update the YDoc within the room, I'm not sure how to broadcast changes out to the clients. This existing code does not appear to broadcast anything to the clients, even when Y.apply_update is called.

doc = Y.YDoc() # server side doc with changes
room = await websocket_server.get_room("my-roomname")
remote_doc = room.ydoc
state_vector = Y.encode_state_vector(remote_doc)
diff = Y.encode_state_as_update(doc, state_vector)
Y.apply_update(remote_doc, diff)

See jupyterlab/jupyterlab#12596 (comment): "we could mitigate the issue about out-of-sync documents by generating the source file from the saved updates when JupyterLab opens a document: if the generated source file and the actual source file are different, we delete all updates from the store and recreate them from scratch, effectively loosing all change history but starting over with a synced document".

On my remove-modeldb branch (https://github.com/dmonad/jupyterlab/tree/remove-modeldb) I get an error when running the ypy-websocket server. Since I didn't change anything related to ypy-websocket, this might affect the master branch as well.

future: <Task finished name='Task-542' coro=<WebsocketServer.serve() done, defined at /home/dmonad/utils/miniconda3/lib/python3.9/site-packages/ypy_websocket/websocket_server.py:97> exception=TypeError('can only concatenate list (not "bytes") to list')>
Traceback (most recent call last):
  File "/home/dmonad/utils/miniconda3/lib/python3.9/asyncio/tasks.py", line 256, in __step
    result = coro.send(None)
  File "/home/dmonad/utils/miniconda3/lib/python3.9/site-packages/ypy_websocket/websocket_server.py", line 100, in serve
    await sync(room.ydoc, websocket)
  File "/home/dmonad/utils/miniconda3/lib/python3.9/site-packages/ypy_websocket/ydoc.py", line 88, in sync
    msg = create_sync_step1_message(state)
  File "/home/dmonad/utils/miniconda3/lib/python3.9/site-packages/ypy_websocket/yutils.py", line 26, in create_sync_step1_message
    return create_message(data, YMessageType.SYNC_STEP1)
  File "/home/dmonad/utils/miniconda3/lib/python3.9/site-packages/ypy_websocket/yutils.py", line 22, in create_message
    return bytes([YMessageType.SYNC, msg_type] + write_var_uint(len(data)) + data)
TypeError: can only concatenate list (not "bytes") to list

Do you have an idea what this error means? I believe I'm running v0.1.9 (I checked pip show ypy_websocket).

I believe our current approach to reading and writing to/from the sqlite database in the SQLiteYStore is inefficient and could lead to incorrect reads that cause difficult to debug bugs.

The approach we are using currently causes 2 db writes for each update, the first one is (roomid, update) and the second is (roomid, metadata). See https://github.com/y-crdt/ypy-websocket/blob/main/ypy_websocket/ystore.py#L153 for details. These two writes are done asynchronously. In the read phase, we query all rows for the path/room and then read the alternating rows as updates and metadata.

There are a couple of issues with this:

1. By storing the update and metadata as separate rows in the database (rather than as separate columns) we are doubling the number or rows, doubling the number of writes, and adding extra processing logic in python to separate out the update and metadata.
2. Because the writes of the path and update are async, it is possible that other read/writes can lead to the interleaving of the alternating update/metadata writes being broken. Thus, we could end up with (update1, update2, metadata1, metadata2). When this is read, it would end up erroneously treating update2 as metadata1 and metadata1 as update2.
3. We are relying on an assumption that the insert order will be preserved in the read order, which is not the case for sqlite (see https://stackoverflow.com/questions/5674385/sqlite-insertion-order-vs-query-order). If this assumption is broken, we would again have mixed updates and metadata.

To fix this, I propose the following:

1. Write/read each update as a single row with (roomid, update, metadata) columns.
2. If we need to preserve insert order upon read, use ORDERBY ROWID. If we write each update as a single row though, this may not be needed though.

I would like to know if you have any examples of ypy-websocket being used in an API.

We currently skip this test because it hangs (after passing).

I'm packaging your module as an rpm package so I'm using the typical PEP517 based build, install and test cycle used on building packages from non-root account.

• python3 -sBm build -w --no-isolation
• because I'm calling build with --no-isolation I'm using during all processes only locally installed modules
• install .whl file in </install/prefix>
• run pytest with $PYTHONPATH pointing to sitearch and sitelib inside </install/prefix>
• build is performed in env which is cut off from access to the public network (pytest is executed with -m "not network")

Here is pytest output:

+ PYTHONPATH=/home/tkloczko/rpmbuild/BUILDROOT/python-ypy-websocket-0.8.4-2.fc35.x86_64/usr/lib64/python3.8/site-packages:/home/tkloczko/rpmbuild/BUILDROOT/python-ypy-websocket-0.8.4-2.fc35.x86_64/usr/lib/python3.8/site-packages
+ /usr/bin/pytest -ra -m 'not network'
==================================================================================== test session starts ====================================================================================
platform linux -- Python 3.8.16, pytest-7.2.2, pluggy-1.0.0
rootdir: /home/tkloczko/rpmbuild/BUILD/ypy-websocket-0.8.4, configfile: pyproject.toml
plugins: asyncio-0.21.0
asyncio: mode=auto
collected 7 items

tests/test_ypy_yjs.py FF                                                                                                                                                              [ 28%]
tests/test_ystore.py .....                                                                                                                                                            [100%]

========================================================================================= FAILURES ==========================================================================================
_____________________________________________________________________________________ test_ypy_yjs_0[0] _____________________________________________________________________________________

yws_server = <ypy_websocket.websocket_server.WebsocketServer object at 0x7f6816f66880>, yjs_client = <subprocess.Popen object at 0x7f6816f66fa0>

    @pytest.mark.asyncio
    @pytest.mark.parametrize("yjs_client", "0", indirect=True)
    async def test_ypy_yjs_0(yws_server, yjs_client):
        ydoc = Y.YDoc()
        ytest = YTest(ydoc)
        websocket = await connect("ws://127.0.0.1:1234/my-roomname")
        WebsocketProvider(ydoc, websocket)
        ymap = ydoc.get_map("map")
        # set a value in "in"
        for v_in in range(10):
            with ydoc.begin_transaction() as t:
                ymap.set(t, "in", float(v_in))
            ytest.run_clock()
>           await ytest.clock_run()

tests/test_ypy_yjs.py:51:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
tests/test_ypy_yjs.py:34: in clock_run
    await asyncio.wait_for(change.wait(), timeout=self.timeout)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

fut = <Task cancelled name='Task-16' coro=<Event.wait() done, defined at /usr/lib64/python3.8/asyncio/locks.py:296>>, timeout = 1.0

    async def wait_for(fut, timeout, *, loop=None):
        """Wait for the single Future or coroutine to complete, with timeout.

        Coroutine will be wrapped in Task.

        Returns result of the Future or coroutine.  When a timeout occurs,
        it cancels the task and raises TimeoutError.  To avoid the task
        cancellation, wrap it in shield().

        If the wait is cancelled, the task is also cancelled.

        This function is a coroutine.
        """
        if loop is None:
            loop = events.get_running_loop()
        else:
            warnings.warn("The loop argument is deprecated since Python 3.8, "
                          "and scheduled for removal in Python 3.10.",
                          DeprecationWarning, stacklevel=2)

        if timeout is None:
            return await fut

        if timeout <= 0:
            fut = ensure_future(fut, loop=loop)

            if fut.done():
                return fut.result()

            await _cancel_and_wait(fut, loop=loop)
            try:
                fut.result()
            except exceptions.CancelledError as exc:
                raise exceptions.TimeoutError() from exc
            else:
                raise exceptions.TimeoutError()

        waiter = loop.create_future()
        timeout_handle = loop.call_later(timeout, _release_waiter, waiter)
        cb = functools.partial(_release_waiter, waiter)

        fut = ensure_future(fut, loop=loop)
        fut.add_done_callback(cb)

        try:
            # wait until the future completes or the timeout
            try:
                await waiter
            except exceptions.CancelledError:
                if fut.done():
                    return fut.result()
                else:
                    fut.remove_done_callback(cb)
                    # We must ensure that the task is not running
                    # after wait_for() returns.
                    # See https://bugs.python.org/issue32751
                    await _cancel_and_wait(fut, loop=loop)
                    raise

            if fut.done():
                return fut.result()
            else:
                fut.remove_done_callback(cb)
                # We must ensure that the task is not running
                # after wait_for() returns.
                # See https://bugs.python.org/issue32751
                await _cancel_and_wait(fut, loop=loop)
>               raise exceptions.TimeoutError()
E               asyncio.exceptions.TimeoutError

/usr/lib64/python3.8/asyncio/tasks.py:501: TimeoutError
----------------------------------------------------------------------------------- Captured stderr call ------------------------------------------------------------------------------------
node:internal/modules/cjs/loader:1093
  throw err;
  ^

Error: Cannot find module 'yjs'
Require stack:
- /home/tkloczko/rpmbuild/BUILD/ypy-websocket-0.8.4/tests/yjs_client_0.js
    at Module._resolveFilename (node:internal/modules/cjs/loader:1090:15)
    at Module._load (node:internal/modules/cjs/loader:934:27)
    at Module.require (node:internal/modules/cjs/loader:1157:19)
    at require (node:internal/modules/helpers:119:18)
    at Object.<anonymous> (/home/tkloczko/rpmbuild/BUILD/ypy-websocket-0.8.4/tests/yjs_client_0.js:1:11)
    at Module._compile (node:internal/modules/cjs/loader:1275:14)
    at Module._extensions..js (node:internal/modules/cjs/loader:1329:10)
    at Module.load (node:internal/modules/cjs/loader:1133:32)
    at Module._load (node:internal/modules/cjs/loader:972:12)
    at Function.executeUserEntryPoint [as runMain] (node:internal/modules/run_main:83:12) {
  code: 'MODULE_NOT_FOUND',
  requireStack: [
    '/home/tkloczko/rpmbuild/BUILD/ypy-websocket-0.8.4/tests/yjs_client_0.js'
  ]
}

Node.js v19.8.1
_____________________________________________________________________________________ test_ypy_yjs_1[1] _____________________________________________________________________________________

yws_server = <ypy_websocket.websocket_server.WebsocketServer object at 0x7f6816f37040>, yjs_client = <subprocess.Popen object at 0x7f6816ee88b0>

    @pytest.mark.asyncio
    @pytest.mark.parametrize("yjs_client", "1", indirect=True)
    async def test_ypy_yjs_1(yws_server, yjs_client):
        # wait for the JS client to connect
        tt, dt = 0, 0.1
        while True:
            await asyncio.sleep(dt)
            if "/my-roomname" in yws_server.rooms:
                break
            tt += dt
            if tt >= 1:
>               raise RuntimeError("Timeout waiting for client to connect")
E               RuntimeError: Timeout waiting for client to connect

tests/test_ypy_yjs.py:67: RuntimeError
----------------------------------------------------------------------------------- Captured stderr call ------------------------------------------------------------------------------------
node:internal/modules/cjs/loader:1093
  throw err;
  ^

Error: Cannot find module 'yjs'
Require stack:
- /home/tkloczko/rpmbuild/BUILD/ypy-websocket-0.8.4/tests/yjs_client_1.js
    at Module._resolveFilename (node:internal/modules/cjs/loader:1090:15)
    at Module._load (node:internal/modules/cjs/loader:934:27)
    at Module.require (node:internal/modules/cjs/loader:1157:19)
    at require (node:internal/modules/helpers:119:18)
    at Object.<anonymous> (/home/tkloczko/rpmbuild/BUILD/ypy-websocket-0.8.4/tests/yjs_client_1.js:1:11)
    at Module._compile (node:internal/modules/cjs/loader:1275:14)
    at Module._extensions..js (node:internal/modules/cjs/loader:1329:10)
    at Module.load (node:internal/modules/cjs/loader:1133:32)
    at Module._load (node:internal/modules/cjs/loader:972:12)
    at Function.executeUserEntryPoint [as runMain] (node:internal/modules/run_main:83:12) {
  code: 'MODULE_NOT_FOUND',
  requireStack: [
    '/home/tkloczko/rpmbuild/BUILD/ypy-websocket-0.8.4/tests/yjs_client_1.js'
  ]
}

Node.js v19.8.1
================================================================================== short test summary info ==================================================================================
FAILED tests/test_ypy_yjs.py::test_ypy_yjs_0[0] - asyncio.exceptions.TimeoutError
FAILED tests/test_ypy_yjs.py::test_ypy_yjs_1[1] - RuntimeError: Timeout waiting for client to connect
================================================================================ 2 failed, 5 passed in 2.33s ================================================================================

Here is list of installed modules in build env

Package           Version
----------------- --------------
aiofiles          23.1.0
attrs             22.2.0
build             0.10.0
distro            1.8.0
editables         0.3
exceptiongroup    1.0.0
gpg               1.18.0-unknown
hatchling         1.13.0
iniconfig         2.0.0
libcomps          0.1.19
packaging         23.0
pathspec          0.11.0
pip               23.0.1
pluggy            1.0.0
pyproject_hooks   1.0.0
pytest            7.2.2
pytest-asyncio    0.21.0
python-dateutil   2.8.2
rpm               4.17.0
six               1.16.0
tomli             2.0.1
typing_extensions 4.5.0
websockets        10.4
wheel             0.38.4
y-py              0.6.1

The (de)serialization functions in yutils.py are generic and could be used for other transports than WebSockets. In Yjs they are part of the lib0 library.
Should we create a new ypy-lib0 repo in this organization?

cc @dmonad

There is currently no test for the YStores, we need to start testing them.
See also #19.

The following sequence of commands results in test failures for yarn but not npm install:

cd tests/
yarn
cd ..
pytest

Right now we are storing configurable metadata along with each update in the YStore. The metadata is stored as a binary blob in the YStore. On the Jupyter side, we are passing a timestamp in the metadata, but we are not currently using that information. I can see a lot of usage cases for metadata, but in most cases, I believe we may want to do more complex queries of the YStore than the current approach allows. Examples include:

1. Time rewinding capabilities would query for patches before or after a particular timestamp.
2. "Suggest mode" would require us to tag updates with an identifier and then query for that identifier.
3. Tracking changes back to users would require a user metadata and queries related to that.
4. "Tagging" a version of a document would require tagging updates and querying based on those tags.

There is still considerable ambiguity about how all of this would work, and I am hesitant to commit to writing the metadata until we dive into all this and figure it out. It is the case that the current approach is flexible and allows us to add metadata fields without breaking the underlying database or file. It is unclear to me if we should (a) leave it as is until we figure these things out or (b) remove metadata until we do, but wanted to bring this up for discussion.

We are finding situations where it would be useful to order updates by date/time. One example is to clean out or combine updates of older documents. The idea is that a YStore subclass could order by room id and sort by date/time and pick the last entry to find the last time the document was edited. If that is longer than a certain amount of time (say a week) we could delete the history of that document or collapse it into a single update.

The proposal is to add a new date/time column to the database and to compute the date/time in the write method.

When using a yjs-websocket server (https://github.com/yjs/y-websocket) the ypy-websocket provider does not sync properly:

thread '<unnamed>' panicked at 'called `Option::unwrap()` on a `None` value', /root/.cargo/registry/src/github.com-1ecc6299db9ec823/yrs-0.11.1/src/block.rs:1073:54

Traceback (most recent call last):
  File "/root/yspine-client/.venv/lib/python3.10/site-packages/ypy_websocket/websocket_provider.py", line 34, in _run
    await process_sync_message(message[1:], self._ydoc, self._websocket, self.log)
  File "/root/yspine-client/.venv/lib/python3.10/site-packages/ypy_websocket/yutils.py", line 130, in process_sync_message
    Y.apply_update(ydoc, update)
pyo3_runtime.PanicException: called `Option::unwrap()` on a `None` value

Other yjs clients on the same server work fine.
Please let me know if you need further details or something to reproduce.

Cheers,
chwzr

Could you please make a new release that includes the fix for #66?

Currently, only the raw Y updates are written to a YStore. We should add more information:

• a timestamp, which can be useful to create a timeline of a document's changes.
• who made the change? That is very application-specific (some applications may not even have a notion of users).
• additional/optional data.

The latter point makes me think that a very generic solution would be for the YStore's encode_state_as_update method to accept a metadata argument. That way, the user is entirely responsible for storing additional data or not, and if so, for choosing a data structure. We could just require that it is e.g. JSON-encodable.

same as y-crdt/ypy#121

Copying and running the code from the README yields the following error:

connection handler failed
Traceback (most recent call last):
  File "/opt/homebrew/Caskroom/miniforge/base/envs/ypy/lib/python3.8/site-packages/websockets/legacy/protocol.py", line 945, in transfer_data
    message = await self.read_message()
  File "/opt/homebrew/Caskroom/miniforge/base/envs/ypy/lib/python3.8/site-packages/websockets/legacy/protocol.py", line 1015, in read_message
    frame = await self.read_data_frame(max_size=self.max_size)
  File "/opt/homebrew/Caskroom/miniforge/base/envs/ypy/lib/python3.8/site-packages/websockets/legacy/protocol.py", line 1090, in read_data_frame
    frame = await self.read_frame(max_size)
  File "/opt/homebrew/Caskroom/miniforge/base/envs/ypy/lib/python3.8/site-packages/websockets/legacy/protocol.py", line 1145, in read_frame
    frame = await Frame.read(
  File "/opt/homebrew/Caskroom/miniforge/base/envs/ypy/lib/python3.8/site-packages/websockets/legacy/framing.py", line 70, in read
    data = await reader(2)
  File "/opt/homebrew/Caskroom/miniforge/base/envs/ypy/lib/python3.8/asyncio/streams.py", line 721, in readexactly
    raise exceptions.IncompleteReadError(incomplete, n)
asyncio.exceptions.IncompleteReadError: 0 bytes read on a total of 2 expected bytes

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/opt/homebrew/Caskroom/miniforge/base/envs/ypy/lib/python3.8/site-packages/websockets/legacy/server.py", line 232, in handler
    await self.ws_handler(self)
  File "/Users/johnwaidhofer/Desktop/Projects/ypy-websocket/ypy_websocket/websocket_server.py", line 105, in serve
    async for message in websocket:
  File "/opt/homebrew/Caskroom/miniforge/base/envs/ypy/lib/python3.8/site-packages/websockets/legacy/protocol.py", line 482, in __aiter__
    yield await self.recv()
  File "/opt/homebrew/Caskroom/miniforge/base/envs/ypy/lib/python3.8/site-packages/websockets/legacy/protocol.py", line 553, in recv
    await self.ensure_open()
  File "/opt/homebrew/Caskroom/miniforge/base/envs/ypy/lib/python3.8/site-packages/websockets/legacy/protocol.py", line 921, in ensure_open
    raise self.connection_closed_exc()
websockets.exceptions.ConnectionClosedError: no close frame received or sent

System Info

• OS: MacOS Monterey 12.5.1
• Python version: 3.8.12
• Architecture: ARM

Hey there-

I really appreciate the hard work that has gone into this repo thus far.

I'm trying to refactor my ypy-websocket setup with the new Django Channels consumer within my ReactJS + Django application but am having trouble getting things running.

Aside from the initial "connect" I'm unsure if my implementation is correct, how to save an load documents, and it for some reason I am seeing quite a few "disconnect" errors:

WebSocket CONNECT / [127.0.0.1:60208]
WebSocket DISCONNECT / [127.0.0.1:60055]
WebSocket HANDSHAKING / [127.0.0.1:60244]
WebSocket CONNECT / [127.0.0.1:60244]
WebSocket DISCONNECT / [127.0.0.1:60084]
WebSocket HANDSHAKING / [127.0.0.1:60287]
WebSocket CONNECT / [127.0.0.1:60287]

Here's what I have so far from the Django side:

class NoteConsumer(YjsConsumer):

    def make_room_name(self) -> str:
        return self.note_id

    async def make_ydoc(self) -> Y.YDoc:
        doc = Y.YDoc()
        self.document_obj = await get_note(self.note_id)
        if self.document_obj:
            # HOW DO I LOAD THE DOCUMENT IF IT IS SAVED AS JSON IN document_obj.notes_data?
            return doc
        else:
            await self.close()

    async def connect(self) -> None:
        query_string = self.scope['query_string'].decode()
        query_string = parse_qs(query_string)
        params = {key: value[0] if len(
            value) == 1 else value for key, value in query_string.items()}
        user_token = params.get('user_token', None)

        auth_passed = await self.auth_user(user_token)
        if not auth_passed:
            await self.close()
            return

        note_id = params.get('note_id', None)
        self.note_id = note_id.replace('/', '')
        self.room_name = self.make_room_name()
        self.ydoc = await self.make_ydoc()
        await super().connect()


    async def receive(self, text_data=None, bytes_data=None):
        if text_data:
            text = json.loads(text_data)
            request_type = text['type']
            if "document_save" in request_type:
                # IS THIS THE CORRECT WAY TO SAVE THE DOCUMENT TO THE DATABASE?
                doc_text = text.get('document', None)
                doc_data = text.get('bytesDoc', None)
                await update_note(self.document_obj, doc_text, doc_data)
        if bytes_data:
            await self.group_send_message(bytes_data)

    async def auth_user(self, auth_token):
        test_auth = get_user(auth_token)
        if test_auth:
            return True
        else:
            return False

For further context on the data we want to retrieve/save from/to the database, we use Y.encodeStateAsUpdate(doc) on the JS client side, and save as JSON to our Postgres database.

Thanks for your time!!