Fork to maintain patches against the official gitea for https://code.ceondo.com https://github.com/go-gitea/gitea

unique_queue.go 1.7KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071
  1. // Copyright 2016 The Gogs Authors. All rights reserved.
  2. // Use of this source code is governed by a MIT-style
  3. // license that can be found in the LICENSE file.
  4. package sync
  5. import (
  6. "github.com/Unknwon/com"
  7. )
  8. // UniqueQueue is a queue which guarantees only one instance of same
  9. // identity is in the line. Instances with same identity will be
  10. // discarded if there is already one in the line.
  11. //
  12. // This queue is particularly useful for preventing duplicated task
  13. // of same purpose.
  14. type UniqueQueue struct {
  15. table *StatusTable
  16. queue chan string
  17. }
  18. // NewUniqueQueue initializes and returns a new UniqueQueue object.
  19. func NewUniqueQueue(queueLength int) *UniqueQueue {
  20. if queueLength <= 0 {
  21. queueLength = 100
  22. }
  23. return &UniqueQueue{
  24. table: NewStatusTable(),
  25. queue: make(chan string, queueLength),
  26. }
  27. }
  28. // Queue returns channel of queue for retrieving instances.
  29. func (q *UniqueQueue) Queue() <-chan string {
  30. return q.queue
  31. }
  32. // Exist returns true if there is an instance with given identity
  33. // exists in the queue.
  34. func (q *UniqueQueue) Exist(id interface{}) bool {
  35. return q.table.IsRunning(com.ToStr(id))
  36. }
  37. // AddFunc adds new instance to the queue with a custom runnable function,
  38. // the queue is blocked until the function exits.
  39. func (q *UniqueQueue) AddFunc(id interface{}, fn func()) {
  40. if q.Exist(id) {
  41. return
  42. }
  43. idStr := com.ToStr(id)
  44. q.table.lock.Lock()
  45. q.table.pool[idStr] = struct{}{}
  46. if fn != nil {
  47. fn()
  48. }
  49. q.table.lock.Unlock()
  50. q.queue <- idStr
  51. }
  52. // Add adds new instance to the queue.
  53. func (q *UniqueQueue) Add(id interface{}) {
  54. q.AddFunc(id, nil)
  55. }
  56. // Remove removes instance from the queue.
  57. func (q *UniqueQueue) Remove(id interface{}) {
  58. q.table.Stop(com.ToStr(id))
  59. }