Something went wrong. Try again.
sloproxy
Something went wrong. Try again.
2.6 kB · 128 lines
Go
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129package pluginhost
import ( "context" "fmt" "sync")
type guardedPluginClient struct { mu sync.Mutex cond *sync.Cond inner pluginClient calls int closed bool shutdownDone chan struct{}}
func newGuardedPluginClient(inner pluginClient) *guardedPluginClient { client := &guardedPluginClient{inner: inner, shutdownDone: make(chan struct{})} client.cond = sync.NewCond(&client.mu) return client}
func (c *guardedPluginClient) Call(ctx context.Context, method string, request []byte) ([]byte, error) { inner, errAcquire := c.acquire() if errAcquire != nil { return nil, errAcquire } if ctx == nil { ctx = context.Background() } result := make(chan guardedPluginCallResult, 1) go func() { defer c.release() defer func() { if recovered := recover(); recovered != nil { result <- guardedPluginCallResult{recovered: recovered} } }() response, errCall := inner.Call(ctx, method, request) result <- guardedPluginCallResult{response: response, err: errCall} }() select { case callResult := <-result: if callResult.recovered != nil { panic(callResult.recovered) } return callResult.response, callResult.err case <-ctx.Done(): return nil, ctx.Err() }}
type guardedPluginCallResult struct { response []byte err error recovered any}
func (c *guardedPluginClient) acquire() (pluginClient, error) { if c == nil { return nil, fmt.Errorf("plugin client is closed") } c.mu.Lock() defer c.mu.Unlock() if c.closed || c.inner == nil { return nil, fmt.Errorf("plugin client is closed") } c.calls++ return c.inner, nil}
func (c *guardedPluginClient) release() { c.mu.Lock() c.calls-- if c.calls == 0 { c.cond.Broadcast() } c.mu.Unlock()}
func (c *guardedPluginClient) Shutdown() { c.ShutdownContext(context.Background())}
// ShutdownContext detaches the client immediately and waits for active calls only// until ctx is canceled. Detached cleanup continues asynchronously when needed.func (c *guardedPluginClient) ShutdownContext(ctx context.Context) { if c == nil { return } if ctx == nil { ctx = context.Background() }
c.mu.Lock() if c.closed { done := c.shutdownDone c.mu.Unlock() select { case <-done: case <-ctx.Done(): } return } c.closed = true inner := c.inner c.inner = nil done := c.shutdownDone c.mu.Unlock()
go func() { c.mu.Lock() for c.calls > 0 { c.cond.Wait() } c.mu.Unlock() if inner != nil { inner.Shutdown() } close(done) }()
select { case <-done: case <-ctx.Done(): }}