Sfoglia il codice sorgente

reject request if backfeed channel is overloaded

backfeed-full-error
Fusl 2 anni fa
parent
commit
65a98d4392
1 ha cambiato i file con 13 aggiunte e 18 eliminazioni
  1. +13
    -18
      main.go

+ 13
- 18
main.go Vedi File

@@ -65,7 +65,7 @@ func (that *ProjectBackfeedManager) RedisConfigDiffers(new *ProjectRedisConfig)
return that.ProjectConfig.RedisConfig == nil || new == nil || *that.ProjectConfig.RedisConfig != *new return that.ProjectConfig.RedisConfig == nil || new == nil || *that.ProjectConfig.RedisConfig != *new
} }


func (that *ProjectBackfeedManager) PushItem(ctx context.Context, item *BackfeedItem) bool {
func (that *ProjectBackfeedManager) PushItem(ctx context.Context, item *BackfeedItem) error {
//that.Lock.RLock() //that.Lock.RLock()
//defer that.Lock.RUnlock() //defer that.Lock.RUnlock()
//if that.C == nil { //if that.C == nil {
@@ -73,11 +73,13 @@ func (that *ProjectBackfeedManager) PushItem(ctx context.Context, item *Backfeed
//} //}
select { select {
case <-ctx.Done(): case <-ctx.Done():
return false
return ctx.Err()
case <-that.Context.Done(): case <-that.Context.Done():
return false
return fmt.Errorf("backfeed channel closed")
case that.C <- item: case that.C <- item:
return true
return nil
default:
return fmt.Errorf("backfeed channel full")
} }
} }


@@ -424,7 +426,6 @@ func (that *GlobalBackfeedManager) HandleLegacy(res http.ResponseWriter, req *ht
scanner := bufio.NewScanner(req.Body) scanner := bufio.NewScanner(req.Body)
scanner.Split(splitter.Split) scanner.Split(splitter.Split)


var err error
statusCode := http.StatusNoContent statusCode := http.StatusNoContent
n := 0 n := 0
for scanner.Scan() { for scanner.Scan() {
@@ -439,25 +440,19 @@ func (that *GlobalBackfeedManager) HandleLegacy(res http.ResponseWriter, req *ht
SecondaryShard: secondaryShard, SecondaryShard: secondaryShard,
Item: bcopy, Item: bcopy,
} }
ok := projectBackfeedManager.PushItem(req.Context(), item)
if !ok {
err = fmt.Errorf("channel closed")
statusCode = http.StatusServiceUnavailable
break
}
n++
}
if err == nil {
err = scanner.Err()
err := projectBackfeedManager.PushItem(req.Context(), item)
if err != nil { if err != nil {
statusCode = http.StatusBadRequest
WriteResponse(res, http.StatusServiceUnavailable, err)
return
} }
n++
} }
err := scanner.Err()
if err != nil { if err != nil {
WriteResponse(res, statusCode, err) WriteResponse(res, statusCode, err)
} else {
WriteResponse(res, http.StatusOK, fmt.Sprintf("%d items queued for deduplication", n))
return
} }
WriteResponse(res, http.StatusOK, fmt.Sprintf("%d items queued for deduplication", n))
return return
} }




Caricamento…
Annulla
Salva