tmc / grpc-websocket-proxy

A proxy to transparently upgrade grpc-gateway streaming endpoints to use websockets

grpc-websocket-proxy

Wrap your grpc-gateway mux with this helper to expose streaming endpoints over websockets.

On the wire this uses newline-delimited json encoding of the messages.

Usage:

	mux := runtime.NewServeMux()
	opts := []grpc.DialOption{grpc.WithInsecure()}
	if err := echoserver.RegisterEchoServiceHandlerFromEndpoint(ctx, mux, *grpcAddr, opts); err != nil {
		return err
	}
-	http.ListenAndServe(*httpAddr, mux)
+	http.ListenAndServe(*httpAddr, wsproxy.WebsocketProxy(mux))

wsproxy

import "github.com/tmc/grpc-websocket-proxy/wsproxy"

Package wsproxy implements a websocket proxy for grpc-gateway backed services

Usage

var (
	MethodOverrideParam = "method"
	TokenCookieName     = "token"
)

func WebsocketProxy

func WebsocketProxy(h http.Handler) http.HandlerFunc

WebsocketProxy attempts to expose the underlying handler as a bidi websocket stream with newline-delimited JSON as the content encoding.

The HTTP Authorization header is either populated from the Sec-Websocket-Protocol field or by a cookie. The cookie name is specified by the TokenCookieName value.

example:

Sec-Websocket-Protocol: Bearer, foobar

is converted to:

Authorization: Bearer foobar

Method can be overwritten with the MethodOverrideParam get parameter in the requested URL

grpc-websocket-proxy's Issues

Propose upstream integration.

I'd like to get this merged into the gateway as it's very closely related to it.

How to enable debug logging

I use etcd 3.3.15 which has this module compiled in.
I set --debug=true option to enable debug messages, but I don't see messages from this module in log file.
So, how can I enable it?

Client example for understanding the streaming REST APIs

I am unable to find a good example to understand how the streaming works via web sockets.

Any link reference or example code would help a lot.

Thanks in advance.

Doesn't work with grpc-gateway/v2/runtime

Doesn't work with grpc-gateway/v2/runtime

I was unable to start with v2. Is there an example of the work?

The connection attempt ends

{
   "code": 12,
   "message": "Method Not Allowed",
   "details": []
}	

Probably a mistake in example

Probably the example service should be.

service EchoService {
  rpc Echo(stream EchoRequest) returns (stream EchoResponse) {
    option (google.api.http) = {get: "/echo"};
  }
  rpc Stream(Empty) returns (stream EchoResponse) {
    option (google.api.http) = {get: "/stream"};
  }
  rpc Heartbeats(stream Empty) returns (stream Heartbeat) {
    option (google.api.http) = {post: "/heartbeats"};
  }
}

Because now both Echo and Stream methods points to the same /echo location. Also the echo should probably be get not post based on what i have tested with JS example.

I could have make a PR, but I don't know which versions you use to generate.

Support non-streaming grpc endpoints

When trying to expose the entire grpc api in websockets, it would be a nice feature to support the request-response endpoints.

srv.Context().Done() is not fired when WebSocket client disconnects?

I'm running into an issue, and I hope you can help me. I'm using your library to stream events published to a Redis server to a WebSocket client (browser).

func (a *DeviceAPI) StreamEventLogs(req *pb.StreamDeviceEventLogsRequest, srv pb.DeviceService_StreamEventLogsServer) error {
	var devEUI lorawan.EUI64

	// For debugging to see when the request has been cancelled
	go func() {
		fmt.Println("cancelled!", <-srv.Context().Done())
	}()

	if err := devEUI.UnmarshalText([]byte(req.DevEui)); err != nil {
		return grpc.Errorf(codes.InvalidArgument, "devEUI: %s", err)
	}

	if err := a.validator.Validate(srv.Context(),
		auth.ValidateNodeAccess(devEUI, auth.Read)); err != nil {
		return grpc.Errorf(codes.Unauthenticated, "authentication failed: %s", err)
	}

	eventLogChan := make(chan eventlog.EventLog)
	go func() {
		err := eventlog.GetEventLogForDevice(srv.Context(), devEUI, eventLogChan)
		if err != nil {
			log.WithError(err).Error("get event-log for device error")
		}
		close(eventLogChan)
	}()

	for el := range eventLogChan {
		b, err := json.Marshal(el.Payload)
		if err != nil {
			return grpc.Errorf(codes.Internal, "marshal json error: %s", err)
		}

		resp := pb.StreamDeviceEventLogsResponse{
			Type:        el.Type,
			PayloadJson: string(b),
		}

		err = srv.Send(&resp)
		if err != nil {
			log.WithError(err).Error("error sending event-log response")
		}
	}

	return nil
}

In short, eventlog.GetEventLogForDevice is executed within a Go routine and blocks until the given context (srv.Context()) is cancelled. This function publishes to eventLogChan. When it completes (with or without error), it closes eventLogChan.

Below there is a loop which reads from eventLogChan and writes this to srv.

---

What I noticed is that when my WebSocket client disconnects, srv.Context().Done() does not fire. So the following code does not print:

	go func() {
		fmt.Println("cancelled!", <-srv.Context().Done())
	}()

Only when srv.Send() is called (e.g. there is a new event to send to the WebSocket client), srv.Context().Done() fires (which will cancel GetEventLogForDevice).

Is this an issue with grpc-websocket-proxy (e.g. the WebSocket disconnect event is not correctly captured), or should I have implemented my code in a different way?

Trying to send large response causes `scanner err: bufio.Scanner: token too long`

When trying to marshal a large response body, the bufio.Scanner returns token too long error here.

WARN[0124] scanner err: bufio.Scanner: token too long                                                     

It would be good to have an option to set a larger buffer size via Scanner.Buffer() to allow reading in larger tokens.

Import logrus as lowercase

go: github.com/tmc/grpc-websocket-proxy/examples/cmd/wsechoserver imports
	github.com/Sirupsen/logrus: github.com/Sirupsen/[email protected]: parsing go.mod:
	module declares its path as: github.com/sirupsen/logrus
	        but was required as: github.com/Sirupsen/logrus

[Bug] Leak

You call io.Pipe() at

grpc-websocket-proxy/wsproxy/websocket_proxy.go

Line 186 in 673ab2c

requestBodyR, requestBodyW := io.Pipe()

then you lose it at

grpc-websocket-proxy/wsproxy/websocket_proxy.go

Line 188 in 673ab2c

if err != nil {

(When error happens at http.NewRequestWithContext you don't close requestBodyW and responseBodyW)

io.Pipe docs
https://pkg.go.dev/io#Pipe

How to send bytes data from client side?

Hi!

I'm trying to send an array byte as a client stream.
When I tried to send the data, the grpc got EOF and never got data from the client side.
In the client side, I'm using Uint8Array as a byte array.(grapc-gateway converts bytes to Uint8Array in javascript)
I'm new to WebSocket, was I missing something?

If you need more detail I will create a minimal reproduction code.

.proto message definition is

message HealthCheckRequest {
    string health = 1;
    bytes buf = 2;
}

client-side is like this.

const byteMessage = new TextEncoder().encode("test message"); // create bytes(Uint8Array) for request
ws.send(JSON.stringify({ health: 'abc', buf: byteMessage})); // this code not working(got EOF)
ws.send(JSON.stringify({ health: 'abc'}));  // without `buf` property work correctly. And I could get data on the server side.

and the server side is like this.

func (s *Server) HealthCheckBiDiStream(stream pb.TranslatorService_HealthCheckBiDiStreamServer) error {
	cnt := 0
	for {
		_, err := stream.Recv()
		if err != nil && err != io.EOF {
			log.Fatal(err)
			return err
		}
		if err == io.EOF {
			return nil
		}

		stream.Send(&pb.HealthCheckResponse{
			Health: fmt.Sprintf("OK %d", cnt),
		})

		if err != nil {
			log.Fatal(err)
			return err
		}
		cnt += 1

	}

}

Add a client example

Either using gopherjs or pure JS, just to illustrate how the MethodOverrideParam interacts with the grpc-gateway exposed gRPC methods. Probably an example of authentication too would be nice.

Passing authentication from browser JS websocket

Hey there, I see there are some docs about passing authorization using the Sec-Websocket-Protocol header. While this is a clever workaround, it doesn't appear as though it'll actually work for websockets started from the browser. From the MDN article on websockets, only registered IANA subprotocols can be passed. Attempting to pass Bearer, 123 results in

VM3291:1 Uncaught DOMException: Failed to construct 'WebSocket': The subprotocol 'Bearer, 123' is invalid.

Was this implemented with browser compatibility in mind? Or was the idea that cookies would be the only browser-friendly authorization method?

Allow client-side streaming communication

Thanks for the work on this proxy! I'm wondering if there is any example of client-side streaming implemented with this proxy. So far, I was able to send all the client-streaming messages to the server perfectly. However, grpc-gateway's generated code expects some EOF before moving on to CloseAndRecv(), so I currently cannot get the server to return the message response.

I'm wondering if there is any examples of triggering the EOF to allow client-side (not bidi) streaming with the server, or if this package would require modification to make this work.

Does not work after upgrading to grpc gateway 1.7.

Works great before upgrading. After upgrading, the client can not receive the streaming response.

Allow specification of acceptable errors

In my personal use expired jwts are common and often ignorable errors. This project should provide a facility to specify if an error is uninteresting.

Unable to read from socket

For the WebSocket example GET on /echo I am unable to read anything from the socket, only able to write back. Reading simply returns an error EOF, when trying to RecvMsg in the gRPC server.

Downgrade logs to Debug level

When deploying this with a grpc-gateway and grpc server it quickly becomes a bit verbose to log every websocket interaction. It would be nicer to have the logs set to debug level.