unique_queue.go 1.7 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970
  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 indentity
  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] = true
  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. }