()
| 98 | parts.append(f"{k}={v[typ]}") |
| 99 | return " ".join(parts) |
| 100 | def poller_thread() -> None: |
| 101 | client = make_client() |
| 102 | streams_client = make_client("dynamodbstreams") |
| 103 | |
| 104 | # Wait for the stream ARN to appear. |
| 105 | stream_arn = None |
| 106 | while not stream_arn and not stop_poller.is_set(): |
| 107 | try: |
| 108 | desc = client.describe_table(TableName=TABLE_NAME) |
| 109 | stream_arn = desc["Table"].get("LatestStreamArn") |
| 110 | except Exception: |
| 111 | pass |
| 112 | if not stream_arn: |
| 113 | time.sleep(0.5) |
| 114 | |
| 115 | if not stream_arn: |
| 116 | return |
| 117 | |
| 118 | log("POLLER", f"Stream ARN: {stream_arn}") |
| 119 | |
| 120 | # Discover shards and get iterators. |
| 121 | shard_iterators: dict[str, str | None] = {} |
| 122 | |
| 123 | def refresh_shards(): |
| 124 | resp = streams_client.describe_stream(StreamArn=stream_arn) |
| 125 | for shard in resp["StreamDescription"]["Shards"]: |
| 126 | sid = shard["ShardId"] |
| 127 | if sid not in shard_iterators: |
| 128 | it = streams_client.get_shard_iterator( |
| 129 | StreamArn=stream_arn, |
| 130 | ShardId=sid, |
| 131 | ShardIteratorType="TRIM_HORIZON", |
| 132 | ) |
| 133 | shard_iterators[sid] = it["ShardIterator"] |
| 134 | log("POLLER", f"Tracking shard {sid}") |
| 135 | |
| 136 | refresh_shards() |
| 137 | |
| 138 | records_seen = 0 |
| 139 | empty_polls = 0 |
| 140 | |
| 141 | while not stop_poller.is_set(): |
| 142 | got_records = False |
| 143 | |
| 144 | for sid in list(shard_iterators.keys()): |
| 145 | it = shard_iterators[sid] |
| 146 | if not it: |
| 147 | continue |
| 148 | try: |
| 149 | resp = streams_client.get_records(ShardIterator=it, Limit=25) |
| 150 | except Exception as e: |
| 151 | log("POLLER", f"Error reading shard {sid}: {e}") |
| 152 | shard_iterators[sid] = None |
| 153 | continue |
| 154 | |
| 155 | shard_iterators[sid] = resp.get("NextShardIterator") |
| 156 | |
| 157 | for rec in resp.get("Records", []): |
nothing calls this directly
no test coverage detected