func processMessages(ctx context.Context, stream io.Reader, handler func(message proto.Message) error) error {
	const stateReadHeader = 1
	const stateReadMessage = 2

	headerBuffer := make([]byte, 4)
	state := stateReadHeader
	var messageBuffer []byte
	for {
		select {
		case <-ctx.Done():
			return nil
		default:
			break
		}

		switch state {
		case stateReadHeader:
			if _, err := io.ReadFull(stream, headerBuffer); err == nil {
				messageLength := binary.LittleEndian.Uint32(headerBuffer)
				messageBuffer = make([]byte, messageLength)
				state = stateReadMessage
			} else {
				return err
			}

		case stateReadMessage:
			if _, err := io.ReadFull(stream, messageBuffer); err == nil {
				state = stateReadHeader
				message, err := readMessage(uint32(len(messageBuffer)), bufio.NewReader(bytes.NewReader(messageBuffer)))
				if err != nil {
					return err
				}
				err = handler(message)
				if err != nil {
					return err
				}
			} else {
				return err
			}
		}
	}
}
// processMessages continuously reads from stream and handles any protobuf messages.
processMessages continuously reads from stream and handles any protobuf messages.
// processMessages continuously reads from stream and handles any protobuf messages .
func processMessages(ctx context.Context, stream io.Reader, handler func(message proto.Message) error) error {
	const stateReadHeader = 1
	const stateReadMessage = 2
	headerBuffer := make([]byte, 4)
	state := stateReadHeader
	var messageBuffer []byte
	for {
		select {
		case <-ctx.Done():
			return nil
		default:
			break
		}
		switch state {
		case stateReadHeader:
			if _, err := io.ReadFull(stream, headerBuffer); err == nil {
				messageLength := binary.LittleEndian.Uint32(headerBuffer)
				messageBuffer = make([]byte, messageLength)
				state = stateReadMessage
			} else {
				return err
			}
		case stateReadMessage:
			if _, err := io.ReadFull(stream, messageBuffer); err == nil {
				state = stateReadHeader
				message, err := readMessage(uint32(len(messageBuffer)), bufio.NewReader(bytes.NewReader(messageBuffer)))
				if err != nil {
					return err
				}
				err = handler(message)
				if err != nil {
					return err
				}
			} else {
				return err
			}
		}
	}
}
func processMessages ( ctx context . Context , stream io . Reader , handler func ( message proto . Message ) error ) error { const stateReadHeader = 1 const stateReadMessage = 2 headerBuffer := make ( [ ] byte , 4 ) state := stateReadHeader var messageBuffer [ ] byte for { select { case <- ctx . Done ( ) : return nil default : break } switch state { case stateReadHeader : if _ , err := io . ReadFull ( stream , headerBuffer ) ; err == nil { messageLength := binary . LittleEndian . Uint32 ( headerBuffer ) messageBuffer = make ( [ ] byte , messageLength ) state = stateReadMessage } else { return err } case stateReadMessage : if _ , err := io . ReadFull ( stream , messageBuffer ) ; err == nil { state = stateReadHeader message , err := readMessage ( uint32 ( len ( messageBuffer ) ) , bufio . NewReader ( bytes . NewReader ( messageBuffer ) ) ) if err != nil { return err } err = handler ( message ) if err != nil { return err } } else { return err } } } }
processMessages continuously reads from stream and handles any protobuf messages.
processMessages continuously reads from stream and handles any protobuf messages .