* * delete the buckets from the watch manager * and notify the subscribers that the keys are deleted due to deleted buckets * @param deletedbuckets: the buckets to be deleted */
(deletingMessageBucket Message)
| 576 | * @param deletedbuckets: the buckets to be deleted |
| 577 | */ |
| 578 | func (wm *watchManager) deleteBucket(deletingMessageBucket Message) { |
| 579 | bucketName := deletingMessageBucket.BucketName |
| 580 | |
| 581 | if _, ok := wm.lookup[bucketName]; !ok { |
| 582 | return |
| 583 | } |
| 584 | |
| 585 | identifierId := wm.idGenerator.GenId() |
| 586 | victimBucketToSubscribers := make(victimBucketToSubscribers) |
| 587 | victimBucketToSubscribers[identifierId] = wm.lookup[bucketName] |
| 588 | delete(wm.lookup, bucketName) |
| 589 | |
| 590 | // Log before sending to avoid race condition with victimCollector |
| 591 | utils.GetLogger().Printf("[watch_manager] Moving bucket %s to victim channel (identifier: %d)\n", bucketName, identifierId) |
| 592 | |
| 593 | // wait for the victim channel to be available |
| 594 | timeOut := time.After(10 * time.Second) |
| 595 | for { |
| 596 | select { |
| 597 | case <-timeOut: |
| 598 | utils.GetLogger().Printf("[watch_manager] Timeout sending victim bucket %s to channel\n", bucketName) |
| 599 | return |
| 600 | case wm.victimChan <- victimBucketToSubscribers: |
| 601 | // Successfully sent - victimCollector now owns the map, don't access it anymore |
| 602 | return |
| 603 | } |
| 604 | } |
| 605 | } |
| 606 | |
| 607 | // Start starts the watch manager |
| 608 | // Implements Component interface |
no test coverage detected