Leader: stream to client and append each chunk to Redis.
(redis, base_key, source, cache_ttl, lock_ttl)
| 105 | |
| 106 | |
| 107 | def _stream_build(redis, base_key, source, cache_ttl, lock_ttl): |
| 108 | """Leader: stream to client and append each chunk to Redis.""" |
| 109 | chunks_key = _chunks_key(base_key) |
| 110 | status_key = _status_key(base_key) |
| 111 | try: |
| 112 | from django.core.cache import cache as django_cache |
| 113 | |
| 114 | django_cache.delete(base_key) # clear any non-chunked entry under this key |
| 115 | redis.delete(chunks_key, _ready_key(base_key)) |
| 116 | redis.set(status_key, STATUS_BUILDING, ex=lock_ttl) |
| 117 | refresh_interval = max(1, lock_ttl // 4) |
| 118 | last_refresh = 0.0 |
| 119 | for chunk in source(): |
| 120 | redis.rpush(chunks_key, _encode_chunk(chunk)) |
| 121 | now = time.monotonic() |
| 122 | if now - last_refresh >= refresh_interval: |
| 123 | _refresh_build_ttl(redis, base_key, lock_ttl) |
| 124 | last_refresh = now |
| 125 | yield chunk |
| 126 | redis.set(status_key, STATUS_READY) |
| 127 | redis.set(_ready_key(base_key), "1") |
| 128 | redis.expire(chunks_key, cache_ttl) |
| 129 | redis.expire(status_key, cache_ttl) |
| 130 | redis.expire(_ready_key(base_key), cache_ttl) |
| 131 | logger.debug("Cached response in %s chunks", redis.llen(chunks_key)) |
| 132 | except Exception: |
| 133 | logger.exception("Chunk cache build failed for %s", base_key) |
| 134 | redis.delete(chunks_key) |
| 135 | redis.set(status_key, STATUS_ERROR, ex=60) |
| 136 | raise |
| 137 | finally: |
| 138 | redis.delete(_lock_key(base_key)) |
| 139 | |
| 140 | |
| 141 | def _stream_follow(redis, base_key, source, cache_ttl, lock_ttl, poll_interval, max_follower_wait): |
no test coverage detected