1. 1
2. 2
3. 3
4. 4
5. 5
6. 6
7. 7
8. 8
9. 9
10. 10
11. 11
12. 12
13. 13
14. 14
15. 15
16. 16
17. 17
18. 18
19. 19
20. 20
21. 21
22. 22
23. 23
24. 24
25. 25
26. 26
27. 27
28. 28
29. 29
30. 30
31. 31
32. 32
33. 33
34. 34
35. 35
36. 36
37. 37
38. 38
39. 39
40. 40
41. 41
42. 42
43. 43
44. 44
45. 45
46. 46
47. 47
48. 48
49. 49
50. 50
51. 51
52. 52
53. 53
54. 54
55. 55
56. 56
57. 57
58. 58
59. 59
60. 60
61. 61
62. 62
63. 63
64. 64
65. 65
66. 66
67. 67
68. 68
69. 69
70. 70
71. 71
72. 72
73. 73
74. 74
75. 75
76. 76
77. 77
78. 78
79. 79
80. 80
81. 81
82. 82
83. 83
84. 84
85. 85
86. 86
87. 87
88. 88
89. 89
90. 90
91. 91
92. 92
93. 93
94. 94
95. 95
96. 96
97. 97
98. 98
99. 99
100. 100
101. 101
102. 102
103. 103
104. 104
105. 105
106. 106
107. 107
108. 108
109. 109
110. 110
111. 111
112. 112
113. 113
114. 114
115. 115
116. 116
117. 117
118. 118
119. 119
120. 120
121. 121
122. 122
123. 123
124. 124
125. 125
126. 126

// Copyright 2019 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT

package queue

import (
	"fmt"
	"strings"

	"code.gitea.io/gitea/modules/json"
	"code.gitea.io/gitea/modules/log"
	"code.gitea.io/gitea/modules/setting"
)

func validType(t string) (Type, error) {
	if len(t) == 0 {
		return PersistableChannelQueueType, nil
	}
	for _, typ := range RegisteredTypes() {
		if t == string(typ) {
			return typ, nil
		}
	}
	return PersistableChannelQueueType, fmt.Errorf("unknown queue type: %s defaulting to %s", t, string(PersistableChannelQueueType))
}

func getQueueSettings(name string) (setting.QueueSettings, []byte) {
	q := setting.GetQueueSettings(name)
	cfg, err := json.Marshal(q)
	if err != nil {
		log.Error("Unable to marshall generic options: %v Error: %v", q, err)
		log.Error("Unable to create queue for %s", name, err)
		return q, []byte{}
	}
	return q, cfg
}

// CreateQueue for name with provided handler and exemplar
func CreateQueue(name string, handle HandlerFunc, exemplar interface{}) Queue {
	q, cfg := getQueueSettings(name)
	if len(cfg) == 0 {
		return nil
	}

	typ, err := validType(q.Type)
	if err != nil {
		log.Error("Invalid type %s provided for queue named %s defaulting to %s", q.Type, name, string(typ))
	}

	returnable, err := NewQueue(typ, handle, cfg, exemplar)
	if q.WrapIfNecessary && err != nil {
		log.Warn("Unable to create queue for %s: %v", name, err)
		log.Warn("Attempting to create wrapped queue")
		returnable, err = NewQueue(WrappedQueueType, handle, WrappedQueueConfiguration{
			Underlying:  typ,
			Timeout:     q.Timeout,
			MaxAttempts: q.MaxAttempts,
			Config:      cfg,
			QueueLength: q.QueueLength,
			Name:        name,
		}, exemplar)
	}
	if err != nil {
		log.Error("Unable to create queue for %s: %v", name, err)
		return nil
	}

	// Sanity check configuration
	if q.Workers == 0 && (q.BoostTimeout == 0 || q.BoostWorkers == 0 || q.MaxWorkers == 0) {
		log.Warn("Queue: %s is configured to be non-scaling and have no workers\n - this configuration is likely incorrect and could cause Gitea to block", q.Name)
		if pausable, ok := returnable.(Pausable); ok {
			log.Warn("Queue: %s is being paused to prevent data-loss, add workers manually and unpause.", q.Name)
			pausable.Pause()
		}
	}

	return returnable
}

// CreateUniqueQueue for name with provided handler and exemplar
func CreateUniqueQueue(name string, handle HandlerFunc, exemplar interface{}) UniqueQueue {
	q, cfg := getQueueSettings(name)
	if len(cfg) == 0 {
		return nil
	}

	if len(q.Type) > 0 && q.Type != "dummy" && q.Type != "immediate" && !strings.HasPrefix(q.Type, "unique-") {
		q.Type = "unique-" + q.Type
	}

	typ, err := validType(q.Type)
	if err != nil || typ == PersistableChannelQueueType {
		typ = PersistableChannelUniqueQueueType
		if err != nil {
			log.Error("Invalid type %s provided for queue named %s defaulting to %s", q.Type, name, string(typ))
		}
	}

	returnable, err := NewQueue(typ, handle, cfg, exemplar)
	if q.WrapIfNecessary && err != nil {
		log.Warn("Unable to create unique queue for %s: %v", name, err)
		log.Warn("Attempting to create wrapped queue")
		returnable, err = NewQueue(WrappedUniqueQueueType, handle, WrappedUniqueQueueConfiguration{
			Underlying:  typ,
			Timeout:     q.Timeout,
			MaxAttempts: q.MaxAttempts,
			Config:      cfg,
			QueueLength: q.QueueLength,
		}, exemplar)
	}
	if err != nil {
		log.Error("Unable to create unique queue for %s: %v", name, err)
		return nil
	}

	// Sanity check configuration
	if q.Workers == 0 && (q.BoostTimeout == 0 || q.BoostWorkers == 0 || q.MaxWorkers == 0) {
		log.Warn("Queue: %s is configured to be non-scaling and have no workers\n - this configuration is likely incorrect and could cause Gitea to block", q.Name)
		if pausable, ok := returnable.(Pausable); ok {
			log.Warn("Queue: %s is being paused to prevent data-loss, add workers manually and unpause.", q.Name)
			pausable.Pause()
		}
	}

	return returnable.(UniqueQueue)
}