Function CopySignalStreamToMessageChannel

pkg/routing/signal.go:191–223

(
	stream psrpc.Stream[SendType, RecvType],
	ch *MessageChannel,
	reader SignalMessageReader[RecvType],
	config config.SignalRelayConfig,
	promSignalSuccess func(),
	promSignalFailure func(),
)

}

func CopySignalStreamToMessageChannel[SendType, RecvType RelaySignalMessage](
	stream psrpc.Stream[SendType, RecvType],
	ch *MessageChannel,
	reader SignalMessageReader[RecvType],
	config config.SignalRelayConfig,
	promSignalSuccess func(),
	promSignalFailure func(),
) error {
	r := &signalMessageReader[SendType, RecvType]{
		reader: reader,
		config: config,
	}
	for msg := range stream.Channel() {
		res, err := r.Read(msg)
		if err != nil {
			promSignalFailure()
			return err
		}

		for _, r := range res {
			if err = ch.WriteMessage(r); err != nil {
				promSignalFailure()
				return err
			}
			promSignalSuccess()
		}

		if msg.GetClose() {
			return stream.Close(nil)
		}
	}
	return stream.Err()
}

// ----------------------------------------