75 lines
2.4 KiB
Go
75 lines
2.4 KiB
Go
package redisactivity
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"time"
|
|
|
|
"proxy-pool/internal/domain/activitypool"
|
|
)
|
|
|
|
var _ activitypool.StateInventoryReader = (*Adapter)(nil)
|
|
|
|
func (a *Adapter) ReadStateInventory(
|
|
ctx context.Context,
|
|
upstreamIDs []string,
|
|
now time.Time,
|
|
) ([]activitypool.StateInventory, error) {
|
|
if ctx == nil {
|
|
return nil, activitypool.ErrInvalidInventory
|
|
}
|
|
if err := ctx.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
if a == nil || now.IsZero() {
|
|
return nil, activitypool.ErrInvalidInventory
|
|
}
|
|
for _, upstreamID := range upstreamIDs {
|
|
if upstreamID == "" {
|
|
return nil, activitypool.ErrInvalidInventory
|
|
}
|
|
}
|
|
if len(upstreamIDs) == 0 {
|
|
return []activitypool.StateInventory{}, nil
|
|
}
|
|
payload, err := json.Marshal(upstreamIDs)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
result, err := runScript(ctx, a.client, statusScript, []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.healthDue, a.keys.healthQueued, a.keys.healthLeases, a.keys.healthTasks,
|
|
a.keys.healthTaskExpiry, a.keys.healthRefTask, a.keys.healthUnhealthy,
|
|
}, now.UnixMilli(), a.options.CleanupLimit, string(payload))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var reply statusScriptReply
|
|
if err := decodeScriptResult(result, &reply); err != nil {
|
|
return nil, err
|
|
}
|
|
if reply.Status == scriptInvalid {
|
|
return nil, activitypool.ErrInvalidInventory
|
|
}
|
|
if reply.Status == scriptUnavailable {
|
|
return nil, invalidScriptReply("expired cleanup is backlogged")
|
|
}
|
|
if reply.Status != scriptOK || len(reply.Inventories) != len(upstreamIDs) {
|
|
return nil, invalidScriptReply("unexpected state inventory reply")
|
|
}
|
|
inventories := make([]activitypool.StateInventory, len(reply.Inventories))
|
|
for index, item := range reply.Inventories {
|
|
if item.UpstreamID != upstreamIDs[index] || item.Fetched < 0 || item.Checking < 0 ||
|
|
item.Available < 0 || item.Suspect < 0 || item.Draining < 0 || item.Unhealthy < 0 || item.Extracted < 0 {
|
|
return nil, invalidScriptReply("invalid state inventory counters")
|
|
}
|
|
inventories[index] = activitypool.StateInventory{
|
|
UpstreamID: item.UpstreamID, Fetched: item.Fetched, Checking: item.Checking,
|
|
Available: item.Available, Suspect: item.Suspect, Draining: item.Draining,
|
|
Unhealthy: item.Unhealthy, Extracted: item.Extracted,
|
|
}
|
|
}
|
|
return inventories, nil
|
|
}
|