feat: Add notification support for bucketCreates and removal (#10075)

cmd/bucket-notification-handlers.go

@@ -17,15 +17,12 @@
 package cmd
 
 import (
-	"encoding/json"
 	"encoding/xml"
 	"io"
 	"net/http"
 	"reflect"
-	"time"
 
 	"github.com/gorilla/mux"
-	xhttp "github.com/minio/minio/cmd/http"
 	"github.com/minio/minio/cmd/logger"
 	"github.com/minio/minio/pkg/bucket/policy"
 	"github.com/minio/minio/pkg/event"
@@ -173,137 +170,3 @@ func (api objectAPIHandlers) PutBucketNotificationHandler(w http.ResponseWriter,
 
 	writeSuccessResponseHeadersOnly(w)
 }
-
-func (api objectAPIHandlers) ListenBucketNotificationHandler(w http.ResponseWriter, r *http.Request) {
-	ctx := newContext(r, w, "ListenBucketNotification")
-
-	defer logger.AuditLog(w, r, "ListenBucketNotification", mustGetClaimsFromToken(r))
-
-	// Validate if bucket exists.
-	objAPI := api.ObjectAPI()
-	if objAPI == nil {
-		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL, guessIsBrowserReq(r))
-		return
-	}
-
-	if !objAPI.IsNotificationSupported() {
-		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrNotImplemented), r.URL, guessIsBrowserReq(r))
-		return
-	}
-
-	if !objAPI.IsListenBucketSupported() {
-		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrNotImplemented), r.URL, guessIsBrowserReq(r))
-		return
-	}
-
-	vars := mux.Vars(r)
-	bucketName := vars["bucket"]
-
-	values := r.URL.Query()
-	values.Set(peerRESTListenBucket, bucketName)
-
-	var prefix string
-	if len(values[peerRESTListenPrefix]) > 1 {
-		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrFilterNamePrefix), r.URL, guessIsBrowserReq(r))
-		return
-	}
-
-	if len(values[peerRESTListenPrefix]) == 1 {
-		if err := event.ValidateFilterRuleValue(values[peerRESTListenPrefix][0]); err != nil {
-			writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
-			return
-		}
-
-		prefix = values[peerRESTListenPrefix][0]
-	}
-
-	var suffix string
-	if len(values[peerRESTListenSuffix]) > 1 {
-		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrFilterNameSuffix), r.URL, guessIsBrowserReq(r))
-		return
-	}
-
-	if len(values[peerRESTListenSuffix]) == 1 {
-		if err := event.ValidateFilterRuleValue(values[peerRESTListenSuffix][0]); err != nil {
-			writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
-			return
-		}
-
-		suffix = values[peerRESTListenSuffix][0]
-	}
-
-	pattern := event.NewPattern(prefix, suffix)
-
-	var eventNames []event.Name
-	for _, s := range values[peerRESTListenEvents] {
-		eventName, err := event.ParseName(s)
-		if err != nil {
-			writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
-			return
-		}
-
-		eventNames = append(eventNames, eventName)
-	}
-
-	if _, err := objAPI.GetBucketInfo(ctx, bucketName); err != nil {
-		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
-		return
-	}
-
-	rulesMap := event.NewRulesMap(eventNames, pattern, event.TargetID{ID: mustGetUUID()})
-
-	w.Header().Set(xhttp.ContentType, "text/event-stream")
-
-	// Listen Publisher and peer-listen-client uses nonblocking send and hence does not wait for slow receivers.
-	// Use buffered channel to take care of burst sends or slow w.Write()
-	listenCh := make(chan interface{}, 4000)
-
-	peers := newPeerRestClients(globalEndpoints)
-
-	globalHTTPListen.Subscribe(listenCh, ctx.Done(), func(evI interface{}) bool {
-		ev, ok := evI.(event.Event)
-		if !ok {
-			return false
-		}
-		if ev.S3.Bucket.Name != values.Get(peerRESTListenBucket) {
-			return false
-		}
-		return rulesMap.MatchSimple(ev.EventName, ev.S3.Object.Key)
-	})
-
-	for _, peer := range peers {
-		if peer == nil {
-			continue
-		}
-		peer.Listen(listenCh, ctx.Done(), values)
-	}
-
-	keepAliveTicker := time.NewTicker(500 * time.Millisecond)
-	defer keepAliveTicker.Stop()
-
-	enc := json.NewEncoder(w)
-	for {
-		select {
-		case evI := <-listenCh:
-			ev := evI.(event.Event)
-			if len(string(ev.EventName)) > 0 {
-				if err := enc.Encode(struct{ Records []event.Event }{[]event.Event{ev}}); err != nil {
-					return
-				}
-			} else {
-				if _, err := w.Write([]byte(" ")); err != nil {
-					return
-				}
-			}
-			w.(http.Flusher).Flush()
-		case <-keepAliveTicker.C:
-			if _, err := w.Write([]byte(" ")); err != nil {
-				return
-			}
-			w.(http.Flusher).Flush()
-		case <-ctx.Done():
-			return
-		}
-	}
-
-}