gate send actions based on whether the peer is connected.
This commit is contained in:
+30
-23
@@ -292,18 +292,20 @@ func (m *RequestManager) Cancel(id string) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
if req.active {
|
if req.active {
|
||||||
go func() {
|
if m.envoy.IsConnected() {
|
||||||
err := m.envoy.Send(envelope.EncloseClose(id))
|
go func() {
|
||||||
if err != nil {
|
err := m.envoy.Send(envelope.EncloseClose(id))
|
||||||
if m.logger != nil {
|
if err != nil {
|
||||||
m.logger.Warn("close send failed", "req", req.id, "error", err)
|
if m.logger != nil {
|
||||||
|
m.logger.Warn("close send failed", "req", req.id, "error", err)
|
||||||
|
}
|
||||||
|
return
|
||||||
}
|
}
|
||||||
return
|
if m.logger != nil {
|
||||||
}
|
m.logger.Debug("close sent", "req", req.id)
|
||||||
if m.logger != nil {
|
}
|
||||||
m.logger.Debug("close sent", "req", req.id)
|
}()
|
||||||
}
|
}
|
||||||
}()
|
|
||||||
req.active = false
|
req.active = false
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -324,7 +326,7 @@ func (m *RequestManager) Close() {
|
|||||||
defer m.mu.Unlock()
|
defer m.mu.Unlock()
|
||||||
|
|
||||||
for id, req := range m.reqs {
|
for id, req := range m.reqs {
|
||||||
if req.active {
|
if m.envoy.IsConnected() && req.active {
|
||||||
go m.envoy.Send(envelope.EncloseClose(id))
|
go m.envoy.Send(envelope.EncloseClose(id))
|
||||||
}
|
}
|
||||||
req.deregisterOnce.Do(func() {
|
req.deregisterOnce.Do(func() {
|
||||||
@@ -336,6 +338,9 @@ func (m *RequestManager) Close() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (m *RequestManager) activate(req *request) {
|
func (m *RequestManager) activate(req *request) {
|
||||||
|
if !m.envoy.IsConnected() {
|
||||||
|
return
|
||||||
|
}
|
||||||
req.active = true
|
req.active = true
|
||||||
go func() {
|
go func() {
|
||||||
err := m.envoy.Send(envelope.EncloseReq(req.id, req.filters))
|
err := m.envoy.Send(envelope.EncloseReq(req.id, req.filters))
|
||||||
@@ -477,18 +482,20 @@ func (m *RequestManager) routeEOSE(msg InboxMessage) {
|
|||||||
req.active = false
|
req.active = false
|
||||||
close(req.buffer)
|
close(req.buffer)
|
||||||
delete(m.reqs, req.id)
|
delete(m.reqs, req.id)
|
||||||
go func() {
|
if m.envoy.IsConnected() {
|
||||||
err := m.envoy.Send(envelope.EncloseClose(subID))
|
go func() {
|
||||||
if err != nil {
|
err := m.envoy.Send(envelope.EncloseClose(subID))
|
||||||
if m.logger != nil {
|
if err != nil {
|
||||||
m.logger.Warn("close send failed", "req", req.id, "error", err)
|
if m.logger != nil {
|
||||||
|
m.logger.Warn("close send failed", "req", req.id, "error", err)
|
||||||
|
}
|
||||||
|
return
|
||||||
}
|
}
|
||||||
return
|
if m.logger != nil {
|
||||||
}
|
m.logger.Debug("close sent", "req", req.id)
|
||||||
if m.logger != nil {
|
}
|
||||||
m.logger.Debug("close sent", "req", req.id)
|
}()
|
||||||
}
|
}
|
||||||
}()
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user