|
|
@ -19,7 +19,6 @@ package cmd |
|
|
|
import ( |
|
|
|
import ( |
|
|
|
"bytes" |
|
|
|
"bytes" |
|
|
|
"encoding/xml" |
|
|
|
"encoding/xml" |
|
|
|
"errors" |
|
|
|
|
|
|
|
"fmt" |
|
|
|
"fmt" |
|
|
|
"net/url" |
|
|
|
"net/url" |
|
|
|
"path" |
|
|
|
"path" |
|
|
@ -112,7 +111,7 @@ func (en *eventNotifier) SetSNSTarget(snsARN string, listenerCh chan []Notificat |
|
|
|
en.rwMutex.Lock() |
|
|
|
en.rwMutex.Lock() |
|
|
|
defer en.rwMutex.Unlock() |
|
|
|
defer en.rwMutex.Unlock() |
|
|
|
if listenerCh == nil { |
|
|
|
if listenerCh == nil { |
|
|
|
return errors.New("invalid argument") |
|
|
|
return errInvalidArgument |
|
|
|
} |
|
|
|
} |
|
|
|
en.snsTargets[snsARN] = append(en.snsTargets[snsARN], listenerCh) |
|
|
|
en.snsTargets[snsARN] = append(en.snsTargets[snsARN], listenerCh) |
|
|
|
return nil |
|
|
|
return nil |
|
|
@ -161,7 +160,7 @@ func (en *eventNotifier) SetBucketNotificationConfig(bucket string, notification |
|
|
|
en.rwMutex.Lock() |
|
|
|
en.rwMutex.Lock() |
|
|
|
defer en.rwMutex.Unlock() |
|
|
|
defer en.rwMutex.Unlock() |
|
|
|
if notificationCfg == nil { |
|
|
|
if notificationCfg == nil { |
|
|
|
return errors.New("invalid argument") |
|
|
|
return errInvalidArgument |
|
|
|
} |
|
|
|
} |
|
|
|
en.notificationConfigs[bucket] = notificationCfg |
|
|
|
en.notificationConfigs[bucket] = notificationCfg |
|
|
|
return nil |
|
|
|
return nil |
|
|
@ -178,8 +177,11 @@ func eventNotify(event eventData) { |
|
|
|
// - s3:ObjectCreated:CompleteMultipartUpload
|
|
|
|
// - s3:ObjectCreated:CompleteMultipartUpload
|
|
|
|
// - s3:ObjectRemoved:Delete
|
|
|
|
// - s3:ObjectRemoved:Delete
|
|
|
|
|
|
|
|
|
|
|
|
nConfig := eventN.GetBucketNotificationConfig(event.Bucket) |
|
|
|
nConfig := globalEventNotifier.GetBucketNotificationConfig(event.Bucket) |
|
|
|
// No bucket notifications enabled, drop the event notification.
|
|
|
|
// No bucket notifications enabled, drop the event notification.
|
|
|
|
|
|
|
|
if nConfig == nil { |
|
|
|
|
|
|
|
return |
|
|
|
|
|
|
|
} |
|
|
|
if len(nConfig.QueueConfigs) == 0 && len(nConfig.TopicConfigs) == 0 && len(nConfig.LambdaConfigs) == 0 { |
|
|
|
if len(nConfig.QueueConfigs) == 0 && len(nConfig.TopicConfigs) == 0 && len(nConfig.LambdaConfigs) == 0 { |
|
|
|
return |
|
|
|
return |
|
|
|
} |
|
|
|
} |
|
|
@ -198,7 +200,7 @@ func eventNotify(event eventData) { |
|
|
|
eventMatch := eventMatch(eventType, qConfig.Events) |
|
|
|
eventMatch := eventMatch(eventType, qConfig.Events) |
|
|
|
ruleMatch := filterRuleMatch(objectName, qConfig.Filter.Key.FilterRules) |
|
|
|
ruleMatch := filterRuleMatch(objectName, qConfig.Filter.Key.FilterRules) |
|
|
|
if eventMatch && ruleMatch { |
|
|
|
if eventMatch && ruleMatch { |
|
|
|
targetLog := eventN.GetQueueTarget(qConfig.QueueARN) |
|
|
|
targetLog := globalEventNotifier.GetQueueTarget(qConfig.QueueARN) |
|
|
|
if targetLog != nil { |
|
|
|
if targetLog != nil { |
|
|
|
targetLog.WithFields(logrus.Fields{ |
|
|
|
targetLog.WithFields(logrus.Fields{ |
|
|
|
"Records": notificationEvent, |
|
|
|
"Records": notificationEvent, |
|
|
@ -211,7 +213,7 @@ func eventNotify(event eventData) { |
|
|
|
ruleMatch := filterRuleMatch(objectName, topicConfig.Filter.Key.FilterRules) |
|
|
|
ruleMatch := filterRuleMatch(objectName, topicConfig.Filter.Key.FilterRules) |
|
|
|
eventMatch := eventMatch(eventType, topicConfig.Events) |
|
|
|
eventMatch := eventMatch(eventType, topicConfig.Events) |
|
|
|
if eventMatch && ruleMatch { |
|
|
|
if eventMatch && ruleMatch { |
|
|
|
targetListeners := eventN.GetSNSTarget(topicConfig.TopicARN) |
|
|
|
targetListeners := globalEventNotifier.GetSNSTarget(topicConfig.TopicARN) |
|
|
|
for _, listener := range targetListeners { |
|
|
|
for _, listener := range targetListeners { |
|
|
|
listener <- notificationEvent |
|
|
|
listener <- notificationEvent |
|
|
|
} |
|
|
|
} |
|
|
@ -352,7 +354,7 @@ func loadAllQueueTargets() (map[string]*logrus.Logger, error) { |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
// Global instance of event notification queue.
|
|
|
|
// Global instance of event notification queue.
|
|
|
|
var eventN *eventNotifier |
|
|
|
var globalEventNotifier *eventNotifier |
|
|
|
|
|
|
|
|
|
|
|
// Initialize event notifier.
|
|
|
|
// Initialize event notifier.
|
|
|
|
func initEventNotifier(objAPI ObjectLayer) error { |
|
|
|
func initEventNotifier(objAPI ObjectLayer) error { |
|
|
@ -373,7 +375,7 @@ func initEventNotifier(objAPI ObjectLayer) error { |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
// Inititalize event notifier queue.
|
|
|
|
// Inititalize event notifier queue.
|
|
|
|
eventN = &eventNotifier{ |
|
|
|
globalEventNotifier = &eventNotifier{ |
|
|
|
rwMutex: &sync.RWMutex{}, |
|
|
|
rwMutex: &sync.RWMutex{}, |
|
|
|
notificationConfigs: configs, |
|
|
|
notificationConfigs: configs, |
|
|
|
queueTargets: queueTargets, |
|
|
|
queueTargets: queueTargets, |
|
|
|