<commit_before>package devastator

import (
	"log"

	"github.com/nbusy/cmap"
	"github.com/nbusy/neptulon/jsonrpc"
)

// Queue is a message queue for queueing and sending messages to users.
type Queue struct {
	conns *cmap.CMap                   // user ID -> conn ID
	route *jsonrpc.Router              // route to send messages through
	reqs  map[string]([]queuedRequest) // user ID -> []queuedRequest
}

// NewQueue creates a new queue object and registers the JSON-RPC router to be used for sending queued messages.
func NewQueue(r *jsonrpc.Router) Queue {
	return Queue{
		conns: cmap.New(),
		route: r,
		reqs:  make(map[string]([]queuedRequest)),
	}
}

// SetConn associates a user with a connection by ID.
// If there are pending messages for the user, they are started to be send immediately.
func (q *Queue) SetConn(userID, connID string) {
	q.conns.Set(userID, connID)
	go q.processQueue(userID) // todo: prevent concurrent runs of processQueue or make []queuedRequest thread-safe
}

// RemoveConn removes a user's associated connection ID.
func (q *Queue) RemoveConn(userID string) {
	q.conns.Delete(userID)
}

// AddRequest queues a request message to be sent to the given user.
func (q *Queue) AddRequest(userID string, method string, params interface{}, resHandler func(ctx *jsonrpc.ResCtx)) error {
	r := queuedRequest{Method: method, Params: params, ResHandler: resHandler}
	if rs, ok := q.reqs[userID]; ok {
		q.reqs[userID] = append(rs, r)
	} else {
		q.reqs[userID] = []queuedRequest{{Method: method, Params: params, ResHandler: resHandler}}
	}

	go q.processQueue(userID) // todo: prevent concurrent runs of processQueue or make []queuedRequest thread-safe
	return nil
}

type queuedRequest struct {
	Method     string
	Params     interface{}
	ResHandler func(ctx *jsonrpc.ResCtx)
}

func (q *Queue) processQueue(userID string) {
	connID, ok := q.conns.GetOk(userID)
	if !ok {
		return
	}

	if reqs, ok := q.reqs[userID]; ok {
		for _, req := range reqs {
			if err := q.route.SendRequest(connID.(string), req.Method, req.Params, req.ResHandler); err != nil {
				log.Fatal(err)
			} else {
				// todo: delete from queue
			}
		}
	}
}<commit_msg>remove processed items from queue<commit_after>package devastator

import (
	"log"

	"github.com/nbusy/cmap"
	"github.com/nbusy/neptulon/jsonrpc"
)

// Queue is a message queue for queueing and sending messages to users.
type Queue struct {
	conns *cmap.CMap                   // user ID -> conn ID
	route *jsonrpc.Router              // route to send messages through
	reqs  map[string]([]queuedRequest) // user ID -> []queuedRequest
}

// NewQueue creates a new queue object and registers the JSON-RPC router to be used for sending queued messages.
func NewQueue(r *jsonrpc.Router) Queue {
	return Queue{
		conns: cmap.New(),
		route: r,
		reqs:  make(map[string]([]queuedRequest)),
	}
}

// SetConn associates a user with a connection by ID.
// If there are pending messages for the user, they are started to be send immediately.
func (q *Queue) SetConn(userID, connID string) {
	q.conns.Set(userID, connID)
	go q.processQueue(userID) // todo: prevent concurrent runs of processQueue or make []queuedRequest thread-safe
}

// RemoveConn removes a user's associated connection ID.
func (q *Queue) RemoveConn(userID string) {
	q.conns.Delete(userID)
}

// AddRequest queues a request message to be sent to the given user.
func (q *Queue) AddRequest(userID string, method string, params interface{}, resHandler func(ctx *jsonrpc.ResCtx)) error {
	r := queuedRequest{Method: method, Params: params, ResHandler: resHandler}
	if rs, ok := q.reqs[userID]; ok {
		q.reqs[userID] = append(rs, r)
	} else {
		q.reqs[userID] = []queuedRequest{{Method: method, Params: params, ResHandler: resHandler}}
	}

	go q.processQueue(userID) // todo: prevent concurrent runs of processQueue or make []queuedRequest thread-safe
	return nil
}

type queuedRequest struct {
	Method     string
	Params     interface{}
	ResHandler func(ctx *jsonrpc.ResCtx)
}

func (q *Queue) processQueue(userID string) {
	connID, ok := q.conns.GetOk(userID)
	if !ok {
		return
	}

	if reqs, ok := q.reqs[userID]; ok {
		for i, req := range reqs {
			if err := q.route.SendRequest(connID.(string), req.Method, req.Params, req.ResHandler); err != nil {
				log.Fatal(err)
			} else {
				reqs, reqs[len(reqs)-1] = append(reqs[:i], reqs[i+1:]...), queuedRequest{} // todo: this might not be needed if function is not a pointer val
			}
		}
	}
}
{'fraction_non_alphanumeric': 0.11926605504587157, 'fraction_numerical': 0.00044752741105392703, 'mean_word_length': 3.705263157894737, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7668397', 'n_tokens_mistral': 1544, 'n_tokens_neox': 1485, 'n_words': 483}