package notify import ( "b612.me/stario" "context" "errors" "net" "time" ) type TransportConn struct { logical *LogicalConn generation uint64 remoteAddr net.Addr // binding pins this view to the physical connection that produced it. A // logical session may replace its transport while an operation is in // flight; using the logical's current binding at that point would cross the // reconnect boundary. binding *transportBinding attached bool hasRuntimeConn bool } const ( transportStreamReadBufferSize = 1024 * 1024 transportPacketReadBufferSize = 64 * 1024 transportFrameMaxPayloadBytes = 64 * 1024 * 1024 ) func streamReadBuffer() []byte { return make([]byte, transportStreamReadBufferSize) } func packetReadBuffer() []byte { return make([]byte, transportPacketReadBufferSize) } func newTransportFrameReader(conn net.Conn, queue *stario.StarQueue) *stario.FrameReader { reader := stario.NewFrameReader(conn, queue) if reader == nil { return nil } if transportStreamReadBufferSize > stario.DefaultFrameReaderBufferSize { reader.SetReadBufferSize(transportStreamReadBufferSize) } return reader } type TransportConnRuntimeSnapshot struct { ClientID string RemoteAddress string BindingOwner string LogicalAlive bool BindingCurrent bool LogicalReason string LogicalError string TransportGeneration uint64 Attached bool HasRuntimeConn bool UsesStreamTransport bool Current bool TransportDetachReason string TransportDetachKind string TransportDetachGeneration uint64 TransportDetachError string TransportDetachedAt time.Time ReattachEligible bool } type transportConnServerSender interface { sendTransport(*TransportConn, TransferMsg) (WaitMsg, error) sendTransportWait(*TransportConn, TransferMsg, time.Duration) (Message, error) } type transportConnServerAPI interface { transportConnServerSender SendCtxTransport(context.Context, *TransportConn, string, MsgVal) (Message, error) SendObjTransport(*TransportConn, string, interface{}) error SendObjCtxTransport(context.Context, *TransportConn, string, interface{}) (Message, error) SendWaitObjTransport(*TransportConn, string, interface{}, time.Duration) (Message, error) SendFileTransport(context.Context, *TransportConn, string) error } type serverUDPTransportRuntimeReader interface { serverUDPListenerSnapshot() *net.UDPConn } var errTransportConnRuntimeSnapshotNil = errors.New("transport conn runtime snapshot target is nil") func (c *ClientConn) clientConnRemoteAddrSnapshot() net.Addr { if c == nil { return nil } return c.clientConnLogicalPeerStateSnapshot().clientAddr } func (c *ClientConn) CurrentTransportConn() *TransportConn { return c.currentTransportConnSnapshot() } func (c *ClientConn) currentTransportConnSnapshot() *TransportConn { if c == nil { return nil } logical := c.LogicalConn() if logical == nil { return nil } return logical.currentTransportConnSnapshot() } func (c *LogicalConn) currentTransportConnSnapshot() *TransportConn { if c == nil { return nil } logical := c remoteAddr := c.RemoteAddr() hasRuntimeConn := c.transportSnapshot() != nil server := c.Server() if server != nil { if reader, ok := server.(serverUDPTransportRuntimeReader); ok && reader.serverUDPListenerSnapshot() != nil { if remoteAddr == nil { return nil } return &TransportConn{ logical: logical, generation: c.transportGenerationSnapshot(), remoteAddr: remoteAddr, binding: c.transportBindingSnapshot(), attached: true, hasRuntimeConn: hasRuntimeConn, } } } if !c.transportAttachedSnapshot() { return nil } return &TransportConn{ logical: logical, generation: c.transportGenerationSnapshot(), remoteAddr: remoteAddr, binding: c.transportBindingSnapshot(), attached: true, hasRuntimeConn: hasRuntimeConn, } } func (t *TransportConn) logicalConnSnapshot() *LogicalConn { if t == nil { return nil } return t.logical } func (t *TransportConn) LogicalConn() *LogicalConn { return t.logicalConnSnapshot() } func (t *TransportConn) ClientID() string { logical := t.logicalConnSnapshot() if logical == nil { return "" } return logical.ID() } func (t *TransportConn) RemoteAddr() net.Addr { if t == nil { return nil } return t.remoteAddr } func (t *TransportConn) TransportGeneration() uint64 { if t == nil { return 0 } return t.generation } func (t *TransportConn) Attached() bool { return t != nil && t.attached } func (t *TransportConn) HasRuntimeConn() bool { return t != nil && t.hasRuntimeConn } func (t *TransportConn) UsesStreamTransport() bool { logical := t.logicalConnSnapshot() if logical == nil { return false } return logical.usesStreamTransportSnapshot() } func (t *TransportConn) IsCurrent() bool { if t == nil { return false } logical := t.logicalConnSnapshot() if logical == nil { return false } current := logical.CurrentTransportConn() if current == nil { return false } else if current.generation != t.generation { return false } else if t.binding != nil { return current.binding == t.binding } return transportConnAddrString(current.remoteAddr) == transportConnAddrString(t.remoteAddr) } // serverTransportBindingSnapshot returns the physical binding pinned by an // explicit transport. The logical fallback preserves compatibility for // package-local callers that construct TransportConn values directly. func serverTransportBindingSnapshot(logical *LogicalConn, transport *TransportConn) *transportBinding { if transport != nil && transport.binding != nil { return transport.binding } if logical == nil { return nil } return logical.transportBindingSnapshot() } // serverTransportBindingSnapshotForConn preserves an inbound socket handoff: // peer attach can transfer the same physical conn from a temporary logical // peer to its stable logical peer. In that case the destination's binding owns // the exact inbound conn and must serialize the reply. A different replacement // conn can never satisfy this identity check, so explicit transport sends stay // pinned to their original binding. func serverTransportBindingSnapshotForConn(logical *LogicalConn, transport *TransportConn, conn net.Conn) *transportBinding { if conn != nil && logical != nil { if current := logical.transportBindingSnapshot(); current != nil && current.connSnapshot() == conn { return current } } return serverTransportBindingSnapshot(logical, transport) } func transportConnAddrString(addr net.Addr) string { if addr == nil { return "" } return addr.String() } func (t *TransportConn) transportScope() string { logical := t.logicalConnSnapshot() if logical == nil { return serverFileDomain + ":unknown" } return serverTransportScopeByGeneration(logical, t.TransportGeneration()) } func (t *TransportConn) deliveryScopes() []string { logical := t.logicalConnSnapshot() if logical == nil { return []string{serverFileDomain + ":unknown"} } base := serverFileScope(logical) transport := t.transportScope() if transport == base { return []string{base} } return []string{transport, base} } func (t *TransportConn) runtimeSnapshot() TransportConnRuntimeSnapshot { snapshot := TransportConnRuntimeSnapshot{ ClientID: t.ClientID(), TransportGeneration: t.TransportGeneration(), Attached: t.Attached(), HasRuntimeConn: t.HasRuntimeConn(), UsesStreamTransport: t.UsesStreamTransport(), Current: t.IsCurrent(), } if addr := t.RemoteAddr(); addr != nil { snapshot.RemoteAddress = addr.String() } if logical := t.logicalConnSnapshot(); logical != nil { diag := snapshotBindingDiagnosticsFromLogical(logical, t, t.TransportGeneration()) snapshot.BindingOwner = diag.BindingOwner snapshot.LogicalAlive = diag.BindingAlive snapshot.BindingCurrent = diag.BindingCurrent snapshot.LogicalReason = diag.BindingReason snapshot.LogicalError = diag.BindingError snapshot.TransportDetachReason = diag.TransportDetachReason snapshot.TransportDetachKind = diag.TransportDetachKind snapshot.TransportDetachGeneration = diag.TransportDetachGeneration snapshot.TransportDetachError = diag.TransportDetachError snapshot.TransportDetachedAt = diag.TransportDetachedAt snapshot.ReattachEligible = diag.ReattachEligible if snapshot.LogicalAlive && snapshot.TransportDetachReason != "" && !snapshot.Current { snapshot.LogicalReason = "" snapshot.LogicalError = "" } } return snapshot } func GetTransportConnRuntimeSnapshot(t *TransportConn) (TransportConnRuntimeSnapshot, error) { if t == nil { return TransportConnRuntimeSnapshot{}, errTransportConnRuntimeSnapshotNil } return t.runtimeSnapshot(), nil } func GetCurrentTransportConnRuntimeSnapshot(c *ClientConn) (TransportConnRuntimeSnapshot, bool, error) { if c == nil { return TransportConnRuntimeSnapshot{}, false, errClientConnRuntimeSnapshotNil } transport := c.CurrentTransportConn() if transport == nil { return TransportConnRuntimeSnapshot{}, false, nil } snapshot, err := GetTransportConnRuntimeSnapshot(transport) if err != nil { return TransportConnRuntimeSnapshot{}, false, err } return snapshot, true, nil } func (t *TransportConn) transportConnServerSenderSnapshot() transportConnServerSender { logical := t.logicalConnSnapshot() if logical == nil { return nil } server := logical.Server() if server == nil { return nil } sender, _ := server.(transportConnServerSender) return sender } func (t *TransportConn) transportConnServerAPISnapshot() transportConnServerAPI { logical := t.logicalConnSnapshot() if logical == nil { return nil } server := logical.Server() if server == nil { return nil } api, _ := server.(transportConnServerAPI) return api } func (t *TransportConn) sendTransfer(msg TransferMsg) (WaitMsg, error) { sender := t.transportConnServerSenderSnapshot() if sender == nil { return WaitMsg{}, transportDetachedErrorForTransport(t) } return sender.sendTransport(t, msg) } func (t *TransportConn) sendTransferWait(msg TransferMsg, timeout time.Duration) (Message, error) { sender := t.transportConnServerSenderSnapshot() if sender == nil { return Message{}, transportDetachedErrorForTransport(t) } return sender.sendTransportWait(t, msg, timeout) } func (t *TransportConn) Send(key string, value MsgVal) error { _, err := t.sendTransfer(TransferMsg{ Key: key, Value: value, Type: MSG_ASYNC, }) return err } func (t *TransportConn) SendWait(key string, value MsgVal, timeout time.Duration) (Message, error) { return t.sendTransferWait(TransferMsg{ Key: key, Value: value, Type: MSG_SYNC_ASK, }, timeout) } func (t *TransportConn) SendCtx(ctx context.Context, key string, value MsgVal) (Message, error) { api := t.transportConnServerAPISnapshot() if api == nil { return Message{}, transportDetachedErrorForTransport(t) } return api.SendCtxTransport(ctx, t, key, value) } func (t *TransportConn) SendObj(key string, value interface{}) error { api := t.transportConnServerAPISnapshot() if api == nil { return transportDetachedErrorForTransport(t) } return api.SendObjTransport(t, key, value) } func (t *TransportConn) SendObjCtx(ctx context.Context, key string, value interface{}) (Message, error) { api := t.transportConnServerAPISnapshot() if api == nil { return Message{}, transportDetachedErrorForTransport(t) } return api.SendObjCtxTransport(ctx, t, key, value) } func (t *TransportConn) SendWaitObj(key string, value interface{}, timeout time.Duration) (Message, error) { api := t.transportConnServerAPISnapshot() if api == nil { return Message{}, transportDetachedErrorForTransport(t) } return api.SendWaitObjTransport(t, key, value, timeout) } func (t *TransportConn) SendFile(ctx context.Context, filePath string) error { api := t.transportConnServerAPISnapshot() if api == nil { return transportDetachedErrorForTransport(t) } return api.SendFileTransport(ctx, t, filePath) }