package notify import ( "context" "net" "time" ) const defaultMessageReplyWriteTimeout = 30 * time.Second const ( MSG_SYS MessageType = iota MSG_SYS_WAIT MSG_SYS_REPLY // Deprecated: legacy RSA key-exchange control message. MSG_KEY_CHANGE MSG_ASYNC MSG_SYNC_ASK MSG_SYNC_REPLY ) type MessageType uint8 type NetType uint8 const ( NET_SERVER NetType = iota NET_CLIENT ) type MsgVal []byte type TransferMsg struct { ID uint64 Key string Value MsgVal Type MessageType } type Message struct { NetType LogicalConn *LogicalConn // Deprecated: ClientConn aliases LogicalConn for compatibility. ClientConn *ClientConn TransportConn *TransportConn ServerConn Client inboundTransportProfile *transportProtectionProfile TransferMsg Time time.Time inboundConn net.Conn } type WaitMsg struct { TransferMsg Time time.Time Reply chan Message scope string //Ctx context.Context } type messageLogicalTransferSender interface { sendLogicalContext(context.Context, *LogicalConn, TransferMsg, time.Duration) (WaitMsg, error) } type messageTransportTransferSender interface { sendTransportContextWithWriteTimeout(context.Context, *TransportConn, TransferMsg, time.Duration) (WaitMsg, error) } type messageInboundTransferSender interface { sendTransferInboundContext(context.Context, *LogicalConn, *TransportConn, net.Conn, *transportProtectionProfile, TransferMsg, time.Duration) error } type messageClientTransferSender interface { sendWithContextTimeout(context.Context, TransferMsg, time.Duration) (WaitMsg, error) } type messageReplyWriteTimeoutProvider interface { ReplyWriteTimeout() time.Duration } func (m *Message) Reply(value MsgVal) (err error) { return m.replyContext(context.Background(), value) } func (m *Message) ReplyCtx(ctx context.Context, value MsgVal) (err error) { if ctx == nil { ctx = context.Background() } return m.replyContext(ctx, value) } func (m *Message) replyContext(ctx context.Context, value MsgVal) (err error) { logical := messageLogicalConnSnapshot(m) transport := messageTransportConnSnapshot(m) writeTimeout := defaultMessageReplyWriteTimeout reply := TransferMsg{ ID: m.ID, Key: m.Key, Value: value, Type: m.Type, } if reply.Type == MSG_SYNC_ASK { reply.Type = MSG_SYNC_REPLY } if reply.Type == MSG_SYS_WAIT { reply.Type = MSG_SYS_REPLY } if m.NetType == NET_SERVER { if logical == nil { return transportDetachedErrorForPeer(nil, transport) } server := logical.Server() if server == nil { return transportDetachedErrorForPeer(logical, transport) } if provider, ok := server.(messageReplyWriteTimeoutProvider); ok { writeTimeout = provider.ReplyWriteTimeout() } if m.inboundConn != nil && logical != nil { sender, _ := server.(messageInboundTransferSender) if sender == nil { return transportDetachedErrorForPeer(logical, transport) } return sender.sendTransferInboundContext(ctx, logical, transport, m.inboundConn, messageInboundTransportProtectionSnapshot(m), reply, writeTimeout) } if transport != nil { sender, _ := server.(messageTransportTransferSender) if sender == nil { return transportDetachedErrorForPeer(logical, transport) } _, err = sender.sendTransportContextWithWriteTimeout(ctx, transport, reply, writeTimeout) return err } sender, _ := server.(messageLogicalTransferSender) if sender == nil { return transportDetachedErrorForPeer(logical, transport) } _, err = sender.sendLogicalContext(ctx, logical, reply, writeTimeout) } if m.NetType == NET_CLIENT { if m.ServerConn == nil { return net.ErrClosed } if sender, ok := m.ServerConn.(messageClientTransferSender); ok { _, err = sender.sendWithContextTimeout(ctx, reply, writeTimeout) } else { _, err = m.ServerConn.send(reply) } } return } func (m *Message) ReplyObj(value interface{}) (err error) { return m.ReplyObjCtx(context.Background(), value) } func (m *Message) ReplyObjCtx(ctx context.Context, value interface{}) (err error) { data, err := encode(value) if err != nil { return err } return m.ReplyCtx(ctx, data) } func hydrateServerMessagePeerFields(message Message) Message { if message.LogicalConn == nil { message.LogicalConn = logicalConnFromClient(message.ClientConn) } if message.LogicalConn == nil && message.TransportConn != nil { message.LogicalConn = message.TransportConn.logicalConnSnapshot() } if message.ClientConn == nil && message.LogicalConn != nil { message.ClientConn = message.LogicalConn.compatClientConn() } if message.TransportConn == nil && message.LogicalConn != nil { message.TransportConn = message.LogicalConn.CurrentTransportConn() } if message.inboundConn != nil && message.inboundTransportProfile == nil && message.LogicalConn != nil { profile := message.LogicalConn.transportProtectionProfileSnapshot() message.inboundTransportProfile = &profile } return message } func messageLogicalConnSnapshot(message *Message) *LogicalConn { if message == nil { return nil } if message.LogicalConn != nil { return message.LogicalConn } return logicalConnFromClient(message.ClientConn) } func messageTransportConnSnapshot(message *Message) *TransportConn { if message == nil { return nil } if message.TransportConn != nil { return message.TransportConn } logical := messageLogicalConnSnapshot(message) if logical == nil { return nil } return logical.CurrentTransportConn() } func messageInboundTransportProtectionSnapshot(message *Message) *transportProtectionProfile { if message == nil { return nil } if message.inboundTransportProfile != nil { return message.inboundTransportProfile } if message.inboundConn == nil { return nil } logical := messageLogicalConnSnapshot(message) if logical == nil { return nil } profile := logical.transportProtectionProfileSnapshot() message.inboundTransportProfile = &profile return message.inboundTransportProfile }