93 lines
2.5 KiB
Go
93 lines
2.5 KiB
Go
package redisactivity
|
|
|
|
import (
|
|
"context"
|
|
"time"
|
|
|
|
"proxy-pool/internal/domain/activitypool"
|
|
)
|
|
|
|
const (
|
|
maintenanceInventory = "inventory"
|
|
maintenanceSweep = "sweep"
|
|
)
|
|
|
|
var (
|
|
_ activitypool.InventoryReader = (*Adapter)(nil)
|
|
_ activitypool.Maintainer = (*Adapter)(nil)
|
|
)
|
|
|
|
func (a *Adapter) Inventory(ctx context.Context, upstreamID string, now time.Time) (activitypool.Inventory, error) {
|
|
result := activitypool.Inventory{UpstreamID: upstreamID}
|
|
if ctx == nil {
|
|
return result, activitypool.ErrInvalidInventory
|
|
}
|
|
if err := ctx.Err(); err != nil {
|
|
return result, err
|
|
}
|
|
if a == nil || upstreamID == "" || now.IsZero() {
|
|
return result, activitypool.ErrInvalidInventory
|
|
}
|
|
reply, err := a.runMaintenance(ctx, maintenanceInventory, now, a.options.CleanupLimit, upstreamID)
|
|
if err != nil {
|
|
return result, err
|
|
}
|
|
if reply.Status == scriptInvalid {
|
|
return result, activitypool.ErrInvalidInventory
|
|
}
|
|
if reply.Status != scriptOK || reply.Count < 0 {
|
|
return result, invalidScriptReply("unexpected inventory reply")
|
|
}
|
|
result.Managed = reply.Count
|
|
return result, nil
|
|
}
|
|
|
|
func (a *Adapter) SweepExpired(ctx context.Context, now time.Time, limit int) (int, error) {
|
|
if ctx == nil {
|
|
return 0, activitypool.ErrInvalidMaintenance
|
|
}
|
|
if err := ctx.Err(); err != nil {
|
|
return 0, err
|
|
}
|
|
if a == nil || now.IsZero() || limit <= 0 {
|
|
return 0, activitypool.ErrInvalidMaintenance
|
|
}
|
|
reply, err := a.runMaintenance(ctx, maintenanceSweep, now, limit, "")
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
if reply.Status == scriptInvalid {
|
|
return 0, activitypool.ErrInvalidMaintenance
|
|
}
|
|
if reply.Status != scriptOK || reply.Count < 0 || reply.Count > limit {
|
|
return 0, invalidScriptReply("unexpected expiry sweep reply")
|
|
}
|
|
return reply.Count, nil
|
|
}
|
|
|
|
func (a *Adapter) runMaintenance(
|
|
ctx context.Context,
|
|
operation string,
|
|
now time.Time,
|
|
limit int,
|
|
upstreamID string,
|
|
) (maintenanceScriptReply, error) {
|
|
operationID, err := newOperationID()
|
|
if err != nil {
|
|
return maintenanceScriptReply{}, err
|
|
}
|
|
result, err := runScript(ctx, a.client, sweepScript, []string{
|
|
a.keys.records, a.keys.unique, a.keys.idkeys, a.keys.expiry, a.keys.available,
|
|
a.keys.inventory, a.keys.stateInventory, a.keys.owners, a.keys.ownerExpiry,
|
|
a.keys.operation(operationID),
|
|
}, operation, now.UnixMilli(), limit, upstreamID, operationTTLMillis(a.options.OperationTTL))
|
|
if err != nil {
|
|
return maintenanceScriptReply{}, err
|
|
}
|
|
var reply maintenanceScriptReply
|
|
if err := decodeScriptResult(result, &reply); err != nil {
|
|
return maintenanceScriptReply{}, err
|
|
}
|
|
return reply, nil
|
|
}
|