notify/client_transport.go

package notify

import (
	"b612.me/stario"
	"context"
	"errors"
	"fmt"
	"net"
	"os"
	"time"
)

func batchedControlEnvelope(env Envelope) bool {
	switch env.Kind {
	case EnvelopeSignal, EnvelopeSignalAck:
		return true
	default:
		return false
	}
}

func writeDeadlineFromTimeout(timeout time.Duration) time.Time {
	if timeout <= 0 {
		return time.Time{}
	}
	return time.Now().Add(timeout)
}

func (c *ClientCommon) sendHeartbeat() error {
	_, err := c.sendWait(TransferMsg{
		ID:    10000,
		Key:   "heartbeat",
		Value: nil,
		Type:  MSG_SYS_WAIT,
	}, time.Second*5)
	return err
}

func (c *ClientCommon) handleHeartbeatResult(err error, failedCount int) (int, bool) {
	return c.handleHeartbeatResultWithSession(c.currentClientSessionEpoch(), err, failedCount)
}

func (c *ClientCommon) handleHeartbeatResultWithSession(epoch uint64, err error, failedCount int) (int, bool) {
	if err == nil {
		c.lastHeartbeat = time.Now().Unix()
		return 0, false
	}
	if c.debugMode {
		fmt.Println("failed to recv heartbeat,timeout!")
	}
	failedCount++
	if failedCount < 3 {
		return failedCount, false
	}
	if c.debugMode {
		fmt.Println("heatbeat failed more than 3 times,stop client")
	}
	if !c.stopClientSessionIfCurrent(epoch, "heartbeat failed more than 3 times", errors.New("heartbeat failed more than 3 times")) {
		return failedCount, true
	}
	return failedCount, true
}

func (c *ClientCommon) readFromTransport() (int, []byte, error) {
	return c.readFromTransportBinding(c.clientTransportBindingSnapshot())
}

func (c *ClientCommon) readFromTransportConn(conn net.Conn) (int, []byte, error) {
	return c.readFromTransportBinding(newTransportBinding(conn, nil))
}

func (c *ClientCommon) readFromTransportBinding(binding *transportBinding) (int, []byte, error) {
	return c.readFromTransportBindingWithBuffer(binding, streamReadBuffer())
}

func (c *ClientCommon) readFromTransportBindingWithBuffer(binding *transportBinding, data []byte) (int, []byte, error) {
	if len(data) == 0 {
		data = streamReadBuffer()
	}
	if binding == nil {
		return 0, data, net.ErrClosed
	}
	conn := binding.connSnapshot()
	if conn == nil {
		return 0, data, net.ErrClosed
	}
	if c.maxReadTimeout.Seconds() != 0 {
		_ = conn.SetReadDeadline(time.Now().Add(c.maxReadTimeout))
	}
	readNum, err := conn.Read(data)
	return readNum, data, err
}

func (c *ClientCommon) handleTransportReadResult(readNum int, data []byte, err error) bool {
	return c.handleTransportReadResultWithSession(c.clientStopContextSnapshot(), c.clientTransportConnSnapshot(), c.clientQueueSnapshot(), readNum, data, err, c.currentClientSessionEpoch())
}

func (c *ClientCommon) handleTransportReadResultWithSession(stopCtx context.Context, conn net.Conn, queue *stario.StarQueue, readNum int, data []byte, err error, epoch uint64) bool {
	return c.handleTransportReadResultWithSessionDispatcher(stopCtx, conn, queue, readNum, data, err, epoch, c.clientInboundDispatcherSnapshot())
}

func (c *ClientCommon) handleTransportReadResultWithSessionDispatcher(stopCtx context.Context, conn net.Conn, queue *stario.StarQueue, readNum int, data []byte, err error, epoch uint64, dispatcher *inboundDispatcher) bool {
	route := c.clientSessionRouteSnapshot()
	if route.binding == nil || route.binding.connSnapshot() != conn || route.binding.queueSnapshot() != queue || route.epoch != epoch {
		route = clientSessionRoute{binding: newTransportBinding(conn, queue), epoch: epoch, sessionStopCtx: stopCtx, transportStopCtx: stopCtx}
	}
	return c.handleTransportReadResultAtRoute(stopCtx, route, readNum, data, err, dispatcher)
}

func (c *ClientCommon) handleTransportReadResultAtRoute(stopCtx context.Context, route clientSessionRoute, readNum int, data []byte, err error, dispatcher *inboundDispatcher) bool {
	binding := route.binding
	queue := binding.queueSnapshot()
	if err == os.ErrDeadlineExceeded {
		if readNum != 0 && queue != nil {
			if !c.pushMessageFastAtRoute(route, queue, data[:readNum], dispatcher) {
				queue.ParseMessage(data[:readNum], route.inboundQueueKey())
			}
		}
		return true
	}
	if err != nil {
		select {
		case <-sessionStopChan(stopCtx):
			c.closeClientTransportBinding(binding)
			return false
		default:
		}
		// An expected shutdown closes the socket, which surfaces here as a
		// read on a closed connection. Only report reads that happen while the
		// session is still supposed to be running.
		if c.showError || c.debugMode {
			fmt.Println("client read error", err)
		}
		c.stopClientSessionIfCurrent(route.epoch, "client read error", err)
		return false
	}
	if queue != nil {
		if !c.pushMessageFastAtRoute(route, queue, data[:readNum], dispatcher) {
			queue.ParseMessage(data[:readNum], route.inboundQueueKey())
		}
	}
	return true
}

func (c *ClientCommon) readTransportPayloadPooled(conn net.Conn, reader *stario.FrameReader) ([]byte, func(), error) {
	if reader == nil {
		return nil, nil, net.ErrClosed
	}
	if conn == nil {
		return nil, nil, net.ErrClosed
	}
	if c.maxReadTimeout.Seconds() != 0 {
		_ = conn.SetReadDeadline(time.Now().Add(c.maxReadTimeout))
	}
	return reader.NextPooled()
}

func (c *ClientCommon) handleTransportPayloadReadResultWithSession(stopCtx context.Context, binding *transportBinding, payload []byte, release func(), err error, epoch uint64, dispatcher *inboundDispatcher) bool {
	route := c.clientSessionRouteSnapshot()
	if route.binding != binding || route.epoch != epoch {
		route = clientSessionRoute{binding: binding, epoch: epoch, sessionStopCtx: stopCtx, transportStopCtx: stopCtx}
	}
	return c.handleTransportPayloadReadResultAtRoute(stopCtx, route, payload, release, err, dispatcher)
}

func (c *ClientCommon) handleTransportPayloadReadResultAtRoute(stopCtx context.Context, route clientSessionRoute, payload []byte, release func(), err error, dispatcher *inboundDispatcher) bool {
	binding := route.binding
	if err == os.ErrDeadlineExceeded {
		return true
	}
	if err != nil {
		if release != nil {
			release()
		}
		select {
		case <-sessionStopChan(stopCtx):
			c.closeClientTransportBinding(binding)
			return false
		default:
		}
		// See handleTransportReadResultAtRoute: shutdown reads are expected.
		if c.showError || c.debugMode {
			fmt.Println("client read error", err)
		}
		c.stopClientSessionIfCurrent(route.epoch, "client read error", err)
		return false
	}
	c.dispatchTransportPayloadFastAtRoute(route, payload, release, dispatcher)
	return true
}

func (c *ClientCommon) dispatchTransportPayloadFast(payload []byte, release func(), dispatcher *inboundDispatcher) {
	c.dispatchTransportPayloadFastAtRoute(c.clientSessionRouteSnapshot(), payload, release, dispatcher)
}

func (c *ClientCommon) dispatchTransportPayloadFastAtRoute(route clientSessionRoute, payload []byte, release func(), dispatcher *inboundDispatcher) {
	if len(payload) == 0 {
		if release != nil {
			release()
		}
		return
	}
	plain, plainRelease, err := c.decryptTransportPayloadPooled(payload, release)
	if err != nil {
		if c.showError || c.debugMode {
			fmt.Println("client decode transport payload error", err)
		}
		return
	}
	if c.tryDispatchBorrowedTransportPlainAtRoute(route, plain, plainRelease) {
		return
	}
	if dispatcher == nil {
		now := time.Now()
		err := c.dispatchInboundTransportPlainAtRoute(route, plain, now)
		if plainRelease != nil {
			plainRelease()
		}
		if err != nil && (c.showError || c.debugMode) {
			fmt.Println("client decode envelope error", err)
		}
		return
	}
	owned := plain
	if plainRelease != nil {
		owned = append([]byte(nil), plain...)
		plainRelease()
	}
	c.wg.Add(1)
	if !dispatcher.DispatchSized(clientInboundDispatchSource(), len(owned), func() {
		defer c.wg.Done()
		now := time.Now()
		if err := c.dispatchInboundTransportPlainAtRoute(route, owned, now); err != nil && (c.showError || c.debugMode) {
			fmt.Println("client decode envelope error", err)
		}
	}) {
		c.wg.Done()
	}
}

func (c *ClientCommon) pushMessageFast(queue *stario.StarQueue, data []byte, dispatcher *inboundDispatcher) bool {
	return c.pushMessageFastAtRoute(c.clientSessionRouteSnapshot(), queue, data, dispatcher)
}

func (c *ClientCommon) pushMessageFastAtRoute(route clientSessionRoute, queue *stario.StarQueue, data []byte, dispatcher *inboundDispatcher) bool {
	if queue == nil || dispatcher == nil || len(data) == 0 {
		return false
	}
	if err := queue.ParseMessageView(data, route.inboundQueueKey(), func(frame stario.FrameView) error {
		c.dispatchTransportPayloadFastAtRoute(route, frame.Payload, nil, dispatcher)
		return nil
	}); err != nil && (c.showError || c.debugMode) {
		fmt.Println("client parse inbound frame error", err)
	}
	return true
}

func (c *ClientCommon) writeToTransport(data []byte) error {
	binding := c.clientTransportBindingSnapshot()
	if binding == nil {
		return net.ErrClosed
	}
	return binding.withConnWriteLock(func(conn net.Conn) error {
		if c.maxWriteTimeout.Seconds() != 0 {
			_ = conn.SetWriteDeadline(time.Now().Add(c.maxWriteTimeout))
		}
		return writeFullToConnUnlocked(conn, data)
	})
}

func (c *ClientCommon) writePayloadToTransport(payload []byte) error {
	return c.writePayloadToTransportContext(context.Background(), payload)
}

func (c *ClientCommon) writePayloadToTransportContext(ctx context.Context, payload []byte) error {
	return c.writePayloadToTransportContextTimeout(ctx, payload, 0)
}

func (c *ClientCommon) writePayloadToTransportContextTimeout(ctx context.Context, payload []byte, writeTimeout time.Duration) error {
	binding := c.clientTransportBindingSnapshot()
	return c.writePayloadToTransportBindingContextTimeout(ctx, binding, payload, writeTimeout)
}

func (c *ClientCommon) writePayloadToTransportBindingContextTimeout(ctx context.Context, binding *transportBinding, payload []byte, writeTimeout time.Duration) error {
	if binding == nil {
		return net.ErrClosed
	}
	queue := binding.queueSnapshot()
	if queue == nil {
		return errClientSessionQueueUnavailable
	}
	writeTimeout = shorterPositiveDuration(c.maxWriteTimeoutSnapshot(), writeTimeout)
	lockAcquired, err := binding.withConnWriteLockContextStopTimeout(ctx, nil, writeTimeout, func(conn net.Conn) error {
		return writeFramedPayloadUnlocked(conn, queue, payload)
	})
	if lockAcquired && err != nil {
		if conn := binding.connSnapshot(); conn != nil && !isPacketTransportConn(conn) {
			binding.closeConn()
		}
	}
	return err
}

func (c *ClientCommon) writeControlPayloadToTransport(ctx context.Context, payload []byte, priority controlPriority) error {
	return c.writeControlPayloadToTransportTimeout(ctx, payload, priority, 0)
}

func (c *ClientCommon) writeControlPayloadToTransportTimeout(ctx context.Context, payload []byte, priority controlPriority, writeTimeout time.Duration) error {
	binding := c.clientTransportBindingSnapshot()
	return c.writeControlPayloadToTransportBindingTimeout(ctx, binding, payload, priority, writeTimeout)
}

func (c *ClientCommon) writeControlPayloadToTransportBindingTimeout(ctx context.Context, binding *transportBinding, payload []byte, priority controlPriority, writeTimeout time.Duration) error {
	if binding == nil {
		return net.ErrClosed
	}
	queue := binding.queueSnapshot()
	if queue == nil {
		return errClientSessionQueueUnavailable
	}
	conn := binding.connSnapshot()
	if conn == nil || isPacketTransportConn(conn) {
		return c.writePayloadToTransportBindingContextTimeout(ctx, binding, payload, writeTimeout)
	}
	sender := binding.controlBatchSenderSnapshot()
	if sender == nil {
		return c.writePayloadToTransportBindingContextTimeout(ctx, binding, payload, writeTimeout)
	}
	return sender.submitContext(ctx, payload, shorterPositiveDuration(c.maxWriteTimeoutSnapshot(), writeTimeout), priority)
}