代码拉取完成,页面将自动刷新
package remote
import (
"runtime"
"sync/atomic"
"github.com/AsynkronIT/protoactor-go/internal/queue/goring"
"github.com/AsynkronIT/protoactor-go/internal/queue/mpsc"
"github.com/AsynkronIT/protoactor-go/log"
"github.com/AsynkronIT/protoactor-go/mailbox"
)
const (
mailboxIdle int32 = iota
mailboxRunning int32 = iota
)
const (
mailboxHasNoMessages int32 = iota
mailboxHasMoreMessages int32 = iota
)
type endpointWriterMailbox struct {
userMailbox *goring.Queue
systemMailbox *mpsc.Queue
schedulerStatus int32
hasMoreMessages int32
invoker mailbox.MessageInvoker
batchSize int
dispatcher mailbox.Dispatcher
suspended bool
}
func (m *endpointWriterMailbox) PostUserMessage(message interface{}) {
// batching mailbox only use the message part
m.userMailbox.Push(message)
m.schedule()
}
func (m *endpointWriterMailbox) PostSystemMessage(message interface{}) {
m.systemMailbox.Push(message)
m.schedule()
}
func (m *endpointWriterMailbox) RegisterHandlers(invoker mailbox.MessageInvoker, dispatcher mailbox.Dispatcher) {
m.invoker = invoker
m.dispatcher = dispatcher
}
func (m *endpointWriterMailbox) Start() {
}
func (m *endpointWriterMailbox) schedule() {
atomic.StoreInt32(&m.hasMoreMessages, mailboxHasMoreMessages) // we have more messages to process
if atomic.CompareAndSwapInt32(&m.schedulerStatus, mailboxIdle, mailboxRunning) {
m.dispatcher.Schedule(m.processMessages)
}
}
func (m *endpointWriterMailbox) processMessages() {
// we are about to start processing messages, we can safely reset the message flag of the mailbox
atomic.StoreInt32(&m.hasMoreMessages, mailboxHasNoMessages)
process:
m.run()
// set mailbox to idle
atomic.StoreInt32(&m.schedulerStatus, mailboxIdle)
// check if there are still messages to process (sent after the message loop ended)
if atomic.SwapInt32(&m.hasMoreMessages, mailboxHasNoMessages) == mailboxHasMoreMessages {
// try setting the mailbox back to running
if atomic.CompareAndSwapInt32(&m.schedulerStatus, mailboxIdle, mailboxRunning) {
goto process
}
}
}
func (m *endpointWriterMailbox) run() {
var msg interface{}
defer func() {
if r := recover(); r != nil {
plog.Info("[ACTOR] Recovering", log.Object("actor", m.invoker), log.Object("reason", r), log.Stack())
m.invoker.EscalateFailure(r, msg)
}
}()
for {
// keep processing system messages until queue is empty
if msg = m.systemMailbox.Pop(); msg != nil {
switch msg.(type) {
case *mailbox.SuspendMailbox:
m.suspended = true
case *mailbox.ResumeMailbox:
m.suspended = false
default:
m.invoker.InvokeSystemMessage(msg)
}
continue
}
// didn't process a system message, so break until we are resumed
if m.suspended {
return
}
var ok bool
if msg, ok = m.userMailbox.PopMany(int64(m.batchSize)); ok {
m.invoker.InvokeUserMessage(msg)
} else {
return
}
runtime.Gosched()
}
}
func endpointWriterMailboxProducer(batchSize, initialSize int) mailbox.Producer {
return func() mailbox.Mailbox {
userMailbox := goring.New(int64(initialSize))
systemMailbox := mpsc.New()
return &endpointWriterMailbox{
userMailbox: userMailbox,
systemMailbox: systemMailbox,
hasMoreMessages: mailboxHasNoMessages,
schedulerStatus: mailboxIdle,
batchSize: batchSize,
}
}
}
此处可能存在不合适展示的内容,页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容,可点击提交进行申诉,我们将尽快为您处理。