123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109 |
- // Copyright 2016 The Gogs Authors. All rights reserved.
- // Copyright 2019 The Gitea Authors. All rights reserved.
- // Use of this source code is governed by a MIT-style
- // license that can be found in the LICENSE file.
-
- package sync
-
- import (
- "github.com/unknwon/com"
- )
-
- // UniqueQueue is a queue which guarantees only one instance of same
- // identity is in the line. Instances with same identity will be
- // discarded if there is already one in the line.
- //
- // This queue is particularly useful for preventing duplicated task
- // of same purpose.
- type UniqueQueue struct {
- table *StatusTable
- queue chan string
- closed chan struct{}
- }
-
- // NewUniqueQueue initializes and returns a new UniqueQueue object.
- func NewUniqueQueue(queueLength int) *UniqueQueue {
- if queueLength <= 0 {
- queueLength = 100
- }
-
- return &UniqueQueue{
- table: NewStatusTable(),
- queue: make(chan string, queueLength),
- closed: make(chan struct{}),
- }
- }
-
- // Close closes this queue
- func (q *UniqueQueue) Close() {
- select {
- case <-q.closed:
- default:
- q.table.lock.Lock()
- select {
- case <-q.closed:
- default:
- close(q.closed)
- }
- q.table.lock.Unlock()
- }
- }
-
- // IsClosed returns a channel that is closed when this Queue is closed
- func (q *UniqueQueue) IsClosed() <-chan struct{} {
- return q.closed
- }
-
- // IDs returns the current ids in the pool
- func (q *UniqueQueue) IDs() []interface{} {
- q.table.lock.Lock()
- defer q.table.lock.Unlock()
- ids := make([]interface{}, 0, len(q.table.pool))
- for id := range q.table.pool {
- ids = append(ids, id)
- }
- return ids
- }
-
- // Queue returns channel of queue for retrieving instances.
- func (q *UniqueQueue) Queue() <-chan string {
- return q.queue
- }
-
- // Exist returns true if there is an instance with given identity
- // exists in the queue.
- func (q *UniqueQueue) Exist(id interface{}) bool {
- return q.table.IsRunning(com.ToStr(id))
- }
-
- // AddFunc adds new instance to the queue with a custom runnable function,
- // the queue is blocked until the function exits.
- func (q *UniqueQueue) AddFunc(id interface{}, fn func()) {
- idStr := com.ToStr(id)
- q.table.lock.Lock()
- if _, ok := q.table.pool[idStr]; ok {
- q.table.lock.Unlock()
- return
- }
- q.table.pool[idStr] = struct{}{}
- if fn != nil {
- fn()
- }
- q.table.lock.Unlock()
- select {
- case <-q.closed:
- return
- case q.queue <- idStr:
- return
- }
- }
-
- // Add adds new instance to the queue.
- func (q *UniqueQueue) Add(id interface{}) {
- q.AddFunc(id, nil)
- }
-
- // Remove removes instance from the queue.
- func (q *UniqueQueue) Remove(id interface{}) {
- q.table.Stop(com.ToStr(id))
- }
|