Get a queue instance based on the storage configuration.
(cls, config: StorageConfig)
| 125 | |
| 126 | @classmethod |
| 127 | def get_queue(cls, config: StorageConfig) -> "QueueBuffer": |
| 128 | """Get a queue instance based on the storage configuration.""" |
| 129 | logger = get_logger(__name__) |
| 130 | if config.replay_buffer.enable: |
| 131 | capacity = config.capacity |
| 132 | logger.info( |
| 133 | f"Using AsyncPriorityQueue with capacity {capacity}, reuse_cooldown_time {config.replay_buffer.reuse_cooldown_time}." |
| 134 | ) |
| 135 | return AsyncPriorityQueue( |
| 136 | capacity=capacity, |
| 137 | reuse_cooldown_time=config.replay_buffer.reuse_cooldown_time, |
| 138 | priority_fn=config.replay_buffer.priority_fn, |
| 139 | priority_fn_args=config.replay_buffer.priority_fn_args, |
| 140 | ) |
| 141 | else: |
| 142 | return AsyncQueue(capacity=config.capacity) |
| 143 | |
| 144 | |
| 145 | class AsyncQueue(asyncio.Queue, QueueBuffer): |
no test coverage detected