notify/msg.go

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
	clientRoute             clientSessionRoute
	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 messageClientRouteTransferSender interface {
	sendWithContextTimeoutAtRoute(context.Context, clientSessionRoute, 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 m.clientRoute.bound() {
			if sender, ok := m.ServerConn.(messageClientRouteTransferSender); ok {
				_, err = sender.sendWithContextTimeoutAtRoute(ctx, m.clientRoute, reply, writeTimeout)
				return err
			}
		}
		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
}