Publishes a WebSocket group message synchronously through Redis. gevent's monkey-patching removes select.epoll, which breaks asyncio event loop creation in threadpool threads. This function replicates channels_redis 4.x group_send directly via a sync Redis client, avoiding asyncio
(group_name, message)
| 417 | |
| 418 | |
| 419 | def _gevent_ws_send(group_name, message): |
| 420 | """ |
| 421 | Publishes a WebSocket group message synchronously through Redis. |
| 422 | |
| 423 | gevent's monkey-patching removes select.epoll, which breaks asyncio event |
| 424 | loop creation in threadpool threads. This function replicates channels_redis |
| 425 | 4.x group_send directly via a sync Redis client, avoiding asyncio entirely. |
| 426 | |
| 427 | Matches channels_redis 4.x defaults: prefix="asgi", expiry=60, |
| 428 | group_expiry=86400, msgpack serializer with 12-byte random prefix. |
| 429 | """ |
| 430 | try: |
| 431 | import msgpack |
| 432 | redis = RedisClient.get_buffer() # decode_responses=False for binary values |
| 433 | |
| 434 | prefix = "asgi" |
| 435 | group_expiry = 86400 |
| 436 | channel_expiry = 60 |
| 437 | rand_len = 12 |
| 438 | |
| 439 | group_key = f"{prefix}:group:{group_name}" |
| 440 | now = time.time() |
| 441 | |
| 442 | redis.zremrangebyscore(group_key, 0, now - group_expiry) |
| 443 | raw = redis.zrange(group_key, 0, -1) |
| 444 | if not raw: |
| 445 | return |
| 446 | |
| 447 | channels = [m.decode('utf-8') if isinstance(m, bytes) else m for m in raw] |
| 448 | |
| 449 | # Group channels by non-local name (prefix up to and including "!") so |
| 450 | # specific channels sharing a prefix share one Redis sorted-set key. |
| 451 | nonlocal_map = {} |
| 452 | for ch in channels: |
| 453 | pos = ch.find("!") |
| 454 | nl = ch[:pos + 1] if pos >= 0 else ch |
| 455 | nonlocal_map.setdefault(nl, []).append(ch) |
| 456 | |
| 457 | pipe = redis.pipeline(transaction=False) |
| 458 | for nl, chs in nonlocal_map.items(): |
| 459 | channel_key = prefix + nl |
| 460 | msg = dict(message) |
| 461 | msg["__asgi_channel__"] = chs |
| 462 | serialized = os.urandom(rand_len) + msgpack.packb(msg) |
| 463 | pipe.zadd(channel_key, {serialized: now}) |
| 464 | pipe.expire(channel_key, channel_expiry) |
| 465 | pipe.execute() |
| 466 | except Exception as e: |
| 467 | logger.warning(f"Failed to send WebSocket update: {e}") |
| 468 | |
| 469 | |
| 470 | def send_websocket_update_sync(group_name, event_type, data): |
no test coverage detected