FlClash/core/message.go

package core

import (
	"encoding/json"
	"time"
)

const (
	messageBatchInterval = 16 * time.Millisecond
	messageBatchSize     = 32
	messageQueueSize     = 256
	messagePriorityBurst = 8
	messageEvictAttempts = 4
)

var (
	stateMessageQueue    = make(chan Message, messageQueueSize)
	priorityMessageQueue = make(chan Message, messageQueueSize)
	bulkMessageQueue     = make(chan Message, messageQueueSize)
)

func init() {
	go runMessageBatcher(stateMessageQueue, priorityMessageQueue, bulkMessageQueue, sendMessageBatch)
}

type messageClass int

const (
	stateMessageClass messageClass = iota
	priorityMessageClass
	bulkMessageClass
	// Shares the bulk queue but never evicts from it: a lookup burst would
	// otherwise push out the request events the connection list is built from.
	yieldingBulkMessageClass
)

func classOfMessage(message Message) messageClass {
	switch message.Type {
	case LoadedMessage, GeoUpdateMessage, RouteChangedMessage:
		return stateMessageClass
	case LogMessage, RequestMessage:
		return bulkMessageClass
	case DnsMessage, RequestTrafficMessage:
		return yieldingBulkMessageClass
	default:
		return priorityMessageClass
	}
}

// Request and DNS events fire per connection and per query for as long as the
// Android service runs, engine or not; unheard, they are not worth encoding.
func sendMessage(messageType MessageType, data any) (queued bool) {
	if !hasEventListener() {
		return false
	}
	message := Message{Type: messageType, Data: data}
	switch classOfMessage(message) {
	case stateMessageClass:
		return enqueueState(stateMessageQueue, message)
	case bulkMessageClass:
		return enqueueLatest(bulkMessageQueue, message)
	case yieldingBulkMessageClass:
		return enqueueState(bulkMessageQueue, message)
	default:
		return enqueueLatest(priorityMessageQueue, message)
	}
}

func enqueueState(queue chan Message, message Message) bool {
	select {
	case queue <- message:
		return true
	default:
		return false
	}
}

func enqueueLatest(queue chan Message, message Message) bool {
	for attempt := 0; attempt < messageEvictAttempts; attempt++ {
		select {
		case queue <- message:
			return true
		default:
		}

		select {
		case <-queue:
		default:
		}
	}
	return false
}

func runMessageBatcher(
	stateMessages <-chan Message,
	priorityMessages <-chan Message,
	bulkMessages <-chan Message,
	send func([]Message),
) {
	timer := time.NewTimer(messageBatchInterval)
	if !timer.Stop() {
		<-timer.C
	}
	defer timer.Stop()

	var deadline <-chan time.Time
	batch := make([]Message, 0, messageBatchSize)

	flush := func() {
		if len(batch) == 0 {
			return
		}
		if !timer.Stop() {
			select {
			case <-timer.C:
			default:
			}
		}
		deadline = nil
		current := batch
		batch = make([]Message, 0, messageBatchSize)
		send(current)
	}
	// A closed queue is set to nil so it never wins a select again.
	take := func(queue *<-chan Message, message Message, open bool) bool {
		if !open {
			*queue = nil
			return false
		}
		if len(batch) == 0 {
			timer.Reset(messageBatchInterval)
			deadline = timer.C
		}
		batch = append(batch, message)
		if len(batch) >= messageBatchSize {
			flush()
		}
		return true
	}

	priorityBurst := 0
	for stateMessages != nil || priorityMessages != nil || bulkMessages != nil {
		select {
		case message, open := <-stateMessages:
			take(&stateMessages, message, open)
			continue
		default:
		}

		// Give bulk events one guaranteed opportunity after a bounded priority
		// burst, while retaining priority preference under ordinary load.
		if priorityBurst >= messagePriorityBurst && bulkMessages != nil {
			priorityBurst = 0
			select {
			case message, open := <-bulkMessages:
				take(&bulkMessages, message, open)
				continue
			default:
			}
		}

		select {
		case message, open := <-priorityMessages:
			if take(&priorityMessages, message, open) {
				priorityBurst++
			}
			continue
		default:
		}

		select {
		case message, open := <-stateMessages:
			take(&stateMessages, message, open)
		case message, open := <-priorityMessages:
			if take(&priorityMessages, message, open) {
				priorityBurst++
			}
		case message, open := <-bulkMessages:
			take(&bulkMessages, message, open)
			priorityBurst = 0
		case <-deadline:
			flush()
		}
	}
	flush()
}

type messageBatchCall struct {
	Method    CoreMethod `json:"method"`
	Arguments []Message  `json:"arguments"`
}

func sendMessageBatch(messages []Message) {
	data, err := json.Marshal(messageBatchCall{
		Method:    messageMethod,
		Arguments: messages,
	})
	if err != nil {
		logError("Message batch marshal error: %v", err)
		return
	}
	deliverEvent(data)
}