You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
363 lines
7.7 KiB
363 lines
7.7 KiB
/*
|
|
* Minio Cloud Storage, (C) 2017 Minio, Inc.
|
|
*
|
|
* Licensed under the Apache License, Version 2.0 (the "License");
|
|
* you may not use this file except in compliance with the License.
|
|
* You may obtain a copy of the License at
|
|
*
|
|
* http://www.apache.org/licenses/LICENSE-2.0
|
|
*
|
|
* Unless required by applicable law or agreed to in writing, software
|
|
* distributed under the License is distributed on an "AS IS" BASIS,
|
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
* See the License for the specific language governing permissions and
|
|
* limitations under the License.
|
|
*/
|
|
|
|
package cmd
|
|
|
|
import (
|
|
"fmt"
|
|
"sync"
|
|
)
|
|
|
|
// Notifier represents collection of supported notification queues.
|
|
type notifier struct {
|
|
sync.RWMutex
|
|
AMQP amqpConfigs `json:"amqp"`
|
|
NATS natsConfigs `json:"nats"`
|
|
ElasticSearch elasticSearchConfigs `json:"elasticsearch"`
|
|
Redis redisConfigs `json:"redis"`
|
|
PostgreSQL postgreSQLConfigs `json:"postgresql"`
|
|
Kafka kafkaConfigs `json:"kafka"`
|
|
Webhook webhookConfigs `json:"webhook"`
|
|
MySQL mySQLConfigs `json:"mysql"`
|
|
// Add new notification queues.
|
|
}
|
|
|
|
type amqpConfigs map[string]amqpNotify
|
|
|
|
func (a amqpConfigs) Clone() amqpConfigs {
|
|
a2 := make(amqpConfigs, len(a))
|
|
for k, v := range a {
|
|
a2[k] = v
|
|
}
|
|
return a2
|
|
}
|
|
|
|
func (a amqpConfigs) Validate() error {
|
|
for k, v := range a {
|
|
if err := v.Validate(); err != nil {
|
|
return fmt.Errorf("AMQP [%s] configuration invalid: %s", k, err.Error())
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
type natsConfigs map[string]natsNotify
|
|
|
|
func (a natsConfigs) Clone() natsConfigs {
|
|
a2 := make(natsConfigs, len(a))
|
|
for k, v := range a {
|
|
a2[k] = v
|
|
}
|
|
return a2
|
|
}
|
|
|
|
func (a natsConfigs) Validate() error {
|
|
for k, v := range a {
|
|
if err := v.Validate(); err != nil {
|
|
return fmt.Errorf("NATS [%s] configuration invalid: %s", k, err.Error())
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
type elasticSearchConfigs map[string]elasticSearchNotify
|
|
|
|
func (a elasticSearchConfigs) Clone() elasticSearchConfigs {
|
|
a2 := make(elasticSearchConfigs, len(a))
|
|
for k, v := range a {
|
|
a2[k] = v
|
|
}
|
|
return a2
|
|
}
|
|
|
|
func (a elasticSearchConfigs) Validate() error {
|
|
for k, v := range a {
|
|
if err := v.Validate(); err != nil {
|
|
return fmt.Errorf("ElasticSearch [%s] configuration invalid: %s", k, err.Error())
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
type redisConfigs map[string]redisNotify
|
|
|
|
func (a redisConfigs) Clone() redisConfigs {
|
|
a2 := make(redisConfigs, len(a))
|
|
for k, v := range a {
|
|
a2[k] = v
|
|
}
|
|
return a2
|
|
}
|
|
|
|
func (a redisConfigs) Validate() error {
|
|
for k, v := range a {
|
|
if err := v.Validate(); err != nil {
|
|
return fmt.Errorf("Redis [%s] configuration invalid: %s", k, err.Error())
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
type postgreSQLConfigs map[string]postgreSQLNotify
|
|
|
|
func (a postgreSQLConfigs) Clone() postgreSQLConfigs {
|
|
a2 := make(postgreSQLConfigs, len(a))
|
|
for k, v := range a {
|
|
a2[k] = v
|
|
}
|
|
return a2
|
|
}
|
|
|
|
func (a postgreSQLConfigs) Validate() error {
|
|
for k, v := range a {
|
|
if err := v.Validate(); err != nil {
|
|
return fmt.Errorf("PostgreSQL [%s] configuration invalid: %s", k, err.Error())
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
type kafkaConfigs map[string]kafkaNotify
|
|
|
|
func (a kafkaConfigs) Clone() kafkaConfigs {
|
|
a2 := make(kafkaConfigs, len(a))
|
|
for k, v := range a {
|
|
a2[k] = v
|
|
}
|
|
return a2
|
|
}
|
|
|
|
func (a kafkaConfigs) Validate() error {
|
|
for k, v := range a {
|
|
if err := v.Validate(); err != nil {
|
|
return fmt.Errorf("Kafka [%s] configuration invalid: %s", k, err.Error())
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
type webhookConfigs map[string]webhookNotify
|
|
|
|
func (a webhookConfigs) Clone() webhookConfigs {
|
|
a2 := make(webhookConfigs, len(a))
|
|
for k, v := range a {
|
|
a2[k] = v
|
|
}
|
|
return a2
|
|
}
|
|
|
|
func (a webhookConfigs) Validate() error {
|
|
for k, v := range a {
|
|
if err := v.Validate(); err != nil {
|
|
return fmt.Errorf("Webhook [%s] configuration invalid: %s", k, err.Error())
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
type mySQLConfigs map[string]mySQLNotify
|
|
|
|
func (a mySQLConfigs) Clone() mySQLConfigs {
|
|
a2 := make(mySQLConfigs, len(a))
|
|
for k, v := range a {
|
|
a2[k] = v
|
|
}
|
|
return a2
|
|
}
|
|
|
|
func (a mySQLConfigs) Validate() error {
|
|
for k, v := range a {
|
|
if err := v.Validate(); err != nil {
|
|
return fmt.Errorf("MySQL [%s] configuration invalid: %s", k, err.Error())
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (n *notifier) Validate() error {
|
|
if n == nil {
|
|
return nil
|
|
}
|
|
if err := n.AMQP.Validate(); err != nil {
|
|
return err
|
|
}
|
|
if err := n.NATS.Validate(); err != nil {
|
|
return err
|
|
}
|
|
if err := n.ElasticSearch.Validate(); err != nil {
|
|
return err
|
|
}
|
|
if err := n.Redis.Validate(); err != nil {
|
|
return err
|
|
}
|
|
if err := n.PostgreSQL.Validate(); err != nil {
|
|
return err
|
|
}
|
|
if err := n.Kafka.Validate(); err != nil {
|
|
return err
|
|
}
|
|
if err := n.Webhook.Validate(); err != nil {
|
|
return err
|
|
}
|
|
if err := n.MySQL.Validate(); err != nil {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (n *notifier) SetAMQPByID(accountID string, amqpn amqpNotify) {
|
|
n.Lock()
|
|
defer n.Unlock()
|
|
n.AMQP[accountID] = amqpn
|
|
}
|
|
|
|
func (n *notifier) GetAMQP() map[string]amqpNotify {
|
|
n.RLock()
|
|
defer n.RUnlock()
|
|
return n.AMQP.Clone()
|
|
}
|
|
|
|
func (n *notifier) GetAMQPByID(accountID string) amqpNotify {
|
|
n.RLock()
|
|
defer n.RUnlock()
|
|
return n.AMQP[accountID]
|
|
}
|
|
|
|
func (n *notifier) SetNATSByID(accountID string, natsn natsNotify) {
|
|
n.Lock()
|
|
defer n.Unlock()
|
|
n.NATS[accountID] = natsn
|
|
}
|
|
|
|
func (n *notifier) GetNATS() map[string]natsNotify {
|
|
n.RLock()
|
|
defer n.RUnlock()
|
|
return n.NATS.Clone()
|
|
}
|
|
|
|
func (n *notifier) GetNATSByID(accountID string) natsNotify {
|
|
n.RLock()
|
|
defer n.RUnlock()
|
|
return n.NATS[accountID]
|
|
}
|
|
|
|
func (n *notifier) SetElasticSearchByID(accountID string, es elasticSearchNotify) {
|
|
n.Lock()
|
|
defer n.Unlock()
|
|
n.ElasticSearch[accountID] = es
|
|
}
|
|
|
|
func (n *notifier) GetElasticSearchByID(accountID string) elasticSearchNotify {
|
|
n.RLock()
|
|
defer n.RUnlock()
|
|
return n.ElasticSearch[accountID]
|
|
}
|
|
|
|
func (n *notifier) GetElasticSearch() map[string]elasticSearchNotify {
|
|
n.RLock()
|
|
defer n.RUnlock()
|
|
return n.ElasticSearch.Clone()
|
|
}
|
|
|
|
func (n *notifier) SetRedisByID(accountID string, r redisNotify) {
|
|
n.Lock()
|
|
defer n.Unlock()
|
|
n.Redis[accountID] = r
|
|
}
|
|
|
|
func (n *notifier) GetRedis() map[string]redisNotify {
|
|
n.RLock()
|
|
defer n.RUnlock()
|
|
return n.Redis.Clone()
|
|
}
|
|
|
|
func (n *notifier) GetRedisByID(accountID string) redisNotify {
|
|
n.RLock()
|
|
defer n.RUnlock()
|
|
return n.Redis[accountID]
|
|
}
|
|
|
|
func (n *notifier) GetWebhook() map[string]webhookNotify {
|
|
n.RLock()
|
|
defer n.RUnlock()
|
|
return n.Webhook.Clone()
|
|
}
|
|
|
|
func (n *notifier) GetWebhookByID(accountID string) webhookNotify {
|
|
n.RLock()
|
|
defer n.RUnlock()
|
|
return n.Webhook[accountID]
|
|
}
|
|
|
|
func (n *notifier) SetWebhookByID(accountID string, pgn webhookNotify) {
|
|
n.Lock()
|
|
defer n.Unlock()
|
|
n.Webhook[accountID] = pgn
|
|
}
|
|
|
|
func (n *notifier) SetPostgreSQLByID(accountID string, pgn postgreSQLNotify) {
|
|
n.Lock()
|
|
defer n.Unlock()
|
|
n.PostgreSQL[accountID] = pgn
|
|
}
|
|
|
|
func (n *notifier) GetPostgreSQL() map[string]postgreSQLNotify {
|
|
n.RLock()
|
|
defer n.RUnlock()
|
|
return n.PostgreSQL.Clone()
|
|
}
|
|
|
|
func (n *notifier) GetPostgreSQLByID(accountID string) postgreSQLNotify {
|
|
n.RLock()
|
|
defer n.RUnlock()
|
|
return n.PostgreSQL[accountID]
|
|
}
|
|
|
|
func (n *notifier) SetMySQLByID(accountID string, pgn mySQLNotify) {
|
|
n.Lock()
|
|
defer n.Unlock()
|
|
n.MySQL[accountID] = pgn
|
|
}
|
|
|
|
func (n *notifier) GetMySQL() map[string]mySQLNotify {
|
|
n.RLock()
|
|
defer n.RUnlock()
|
|
return n.MySQL.Clone()
|
|
}
|
|
|
|
func (n *notifier) GetMySQLByID(accountID string) mySQLNotify {
|
|
n.RLock()
|
|
defer n.RUnlock()
|
|
return n.MySQL[accountID]
|
|
}
|
|
|
|
func (n *notifier) SetKafkaByID(accountID string, kn kafkaNotify) {
|
|
n.Lock()
|
|
defer n.Unlock()
|
|
n.Kafka[accountID] = kn
|
|
}
|
|
|
|
func (n *notifier) GetKafka() map[string]kafkaNotify {
|
|
n.RLock()
|
|
defer n.RUnlock()
|
|
return n.Kafka.Clone()
|
|
}
|
|
|
|
func (n *notifier) GetKafkaByID(accountID string) kafkaNotify {
|
|
n.RLock()
|
|
defer n.RUnlock()
|
|
return n.Kafka[accountID]
|
|
}
|
|
|