filer: mysql2, postgres2 trigger actions on bucket creation and deletion

fix #1877

weed/filer/abstract_sql/abstract_sql_store.go

@@ -32,6 +32,25 @@ type AbstractSqlStore struct {
 	dbsLock            sync.Mutex
 }
 
+func (store *AbstractSqlStore) OnBucketCreation(bucket string) {
+	store.dbsLock.Lock()
+	defer store.dbsLock.Unlock()
+
+	if store.dbs == nil {
+		return
+	}
+	store.dbs[bucket] = true
+}
+func (store *AbstractSqlStore) OnBucketDeletion(bucket string) {
+	store.dbsLock.Lock()
+	defer store.dbsLock.Unlock()
+
+	if store.dbs == nil {
+		return
+	}
+	delete(store.dbs, bucket)
+}
+
 const (
 	DEFAULT_TABLE = "filemeta"
 )

weed/filer/filer_on_meta_event.go

@@ -12,6 +12,24 @@ import (
 // onMetadataChangeEvent is triggered after filer processed change events from local or remote filers
 func (f *Filer) onMetadataChangeEvent(event *filer_pb.SubscribeMetadataResponse) {
 	f.maybeReloadFilerConfiguration(event)
+	f.onBucketEvents(event)
+}
+
+func (f *Filer) onBucketEvents(event *filer_pb.SubscribeMetadataResponse) {
+	message := event.EventNotification
+	for _, sig := range message.Signatures {
+		if sig == f.Signature {
+			return
+		}
+	}
+	if f.DirBucketsPath == event.Directory {
+		if message.OldEntry == nil && message.NewEntry != nil {
+			f.Store.OnBucketCreation(message.NewEntry.Name)
+		}
+		if message.OldEntry != nil && message.NewEntry == nil {
+			f.Store.OnBucketDeletion(message.OldEntry.Name)
+		}
+	}
 }
 
 func (f *Filer) maybeReloadFilerConfiguration(event *filer_pb.SubscribeMetadataResponse) {

weed/filer/filerstore.go

@@ -39,3 +39,9 @@ type FilerStore interface {
 
 	Shutdown()
 }
+
+type BucketAware interface {
+	OnBucketCreation(bucket string)
+	OnBucketDeletion(bucket string)
+}
+

weed/filer/filerstore_wrapper.go

@@ -21,6 +21,8 @@ type VirtualFilerStore interface {
 	DeleteHardLink(ctx context.Context, hardLinkId HardLinkId) error
 	DeleteOneEntry(ctx context.Context, entry *Entry) error
 	AddPathSpecificStore(path string, storeId string, store FilerStore)
+	OnBucketCreation(bucket string)
+	OnBucketDeletion(bucket string)
 }
 
 type FilerStoreWrapper struct {
@@ -40,6 +42,27 @@ func NewFilerStoreWrapper(store FilerStore) *FilerStoreWrapper {
 	}
 }
 
+func (fsw *FilerStoreWrapper) OnBucketCreation(bucket string) {
+	for _, store := range fsw.storeIdToStore {
+		if ba, ok := store.(BucketAware); ok {
+			ba.OnBucketCreation(bucket)
+		}
+	}
+	if ba, ok := fsw.defaultStore.(BucketAware); ok {
+		ba.OnBucketCreation(bucket)
+	}
+}
+func (fsw *FilerStoreWrapper) OnBucketDeletion(bucket string) {
+	for _, store := range fsw.storeIdToStore {
+		if ba, ok := store.(BucketAware); ok {
+			ba.OnBucketDeletion(bucket)
+		}
+	}
+	if ba, ok := fsw.defaultStore.(BucketAware); ok {
+		ba.OnBucketDeletion(bucket)
+	}
+}
+
 func (fsw *FilerStoreWrapper) AddPathSpecificStore(path string, storeId string, store FilerStore) {
 	fsw.storeIdToStore[storeId] = NewFilerStorePathTranlator(path, store)
 	err := fsw.pathToStore.Put([]byte(path), storeId)