ae9a71034d
Add HTTP /v1 MVC, TCP send path, groups with server-side member-avatar collage, object-storage upload, and friends/favorites around the architecture-v3 layout. Co-authored-by: Cursor <cursoragent@cursor.com>
118 lines
2.1 KiB
Go
118 lines
2.1 KiB
Go
package model
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"sync"
|
|
"time"
|
|
|
|
"spark-server/internal/module/deliver/entity"
|
|
)
|
|
|
|
type MemoryQueue struct {
|
|
mu sync.Mutex
|
|
queue [][]byte
|
|
processing [][]byte
|
|
wait chan struct{}
|
|
}
|
|
|
|
func NewMemoryQueue() *MemoryQueue {
|
|
return &MemoryQueue{wait: make(chan struct{}, 1)}
|
|
}
|
|
|
|
func (q *MemoryQueue) Push(_ context.Context, job *entity.Job) error {
|
|
raw, err := json.Marshal(job)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
q.mu.Lock()
|
|
q.queue = append(q.queue, raw)
|
|
q.mu.Unlock()
|
|
select {
|
|
case q.wait <- struct{}{}:
|
|
default:
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (q *MemoryQueue) Take(ctx context.Context, timeout time.Duration) (*entity.Job, error) {
|
|
deadline := time.Now().Add(timeout)
|
|
for {
|
|
q.mu.Lock()
|
|
if len(q.queue) > 0 {
|
|
raw := q.queue[0]
|
|
q.queue = q.queue[1:]
|
|
q.processing = append(q.processing, raw)
|
|
q.mu.Unlock()
|
|
job := &entity.Job{}
|
|
if err := json.Unmarshal(raw, job); err != nil {
|
|
return nil, err
|
|
}
|
|
job.SetRaw(raw)
|
|
return job, nil
|
|
}
|
|
q.mu.Unlock()
|
|
remain := time.Until(deadline)
|
|
if remain <= 0 {
|
|
return nil, nil
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
return nil, ctx.Err()
|
|
case <-time.After(remain):
|
|
return nil, nil
|
|
case <-q.wait:
|
|
}
|
|
}
|
|
}
|
|
|
|
func (q *MemoryQueue) Ack(_ context.Context, job *entity.Job) error {
|
|
if job == nil {
|
|
return nil
|
|
}
|
|
raw := job.Raw()
|
|
q.mu.Lock()
|
|
defer q.mu.Unlock()
|
|
for i, p := range q.processing {
|
|
if bytes.Equal(p, raw) {
|
|
q.processing = append(q.processing[:i], q.processing[i+1:]...)
|
|
return nil
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (q *MemoryQueue) RequeueStale(context.Context) error {
|
|
q.mu.Lock()
|
|
defer q.mu.Unlock()
|
|
if len(q.processing) == 0 {
|
|
return nil
|
|
}
|
|
q.queue = append(q.processing, q.queue...)
|
|
q.processing = nil
|
|
return nil
|
|
}
|
|
|
|
type MemoryDone struct {
|
|
mu sync.Mutex
|
|
seen map[int64]struct{}
|
|
}
|
|
|
|
func NewMemoryDone() *MemoryDone {
|
|
return &MemoryDone{seen: make(map[int64]struct{})}
|
|
}
|
|
|
|
func (d *MemoryDone) Mark(_ context.Context, msgID int64) (bool, error) {
|
|
if msgID <= 0 {
|
|
return true, nil
|
|
}
|
|
d.mu.Lock()
|
|
defer d.mu.Unlock()
|
|
if _, ok := d.seen[msgID]; ok {
|
|
return false, nil
|
|
}
|
|
d.seen[msgID] = struct{}{}
|
|
return true, nil
|
|
}
|