(name string)
| 62 | } |
| 63 | |
| 64 | func (f *MemoryQueueFactory) release(name string) { |
| 65 | f.mu.Lock() |
| 66 | defer f.mu.Unlock() |
| 67 | q, ok := f.queues[name] |
| 68 | if !ok { |
| 69 | panic("release non-exist queue: " + name) |
| 70 | } |
| 71 | if atomic.AddInt32(&q.refCnt, -1) == 0 { |
| 72 | close(q.c) |
| 73 | delete(f.queues, name) |
| 74 | } |
| 75 | slog.InfoContext(f.ctx, "Released memory queue", |
| 76 | "current_use_count", atomic.LoadInt32(&q.refCnt), |
| 77 | "name", name) |
| 78 | } |
| 79 | |
| 80 | func (f *MemoryQueueFactory) NewSourceTube(ctx context.Context, configMap ConfigMap) (<-chan Record, error) { |
| 81 | config := SourceQueueConfig{} |
no outgoing calls
no test coverage detected