-
Notifications
You must be signed in to change notification settings - Fork 59
WebSocket proxy for the grpc-gateway #752
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Merged
Merged
Changes from 3 commits
Commits
Show all changes
23 commits
Select commit
Hold shift + click to select a range
7cd3254
Squashed commit of the following:
hariso b51b9d0
send scanner error
hariso 596753a
send scanner error
hariso 9895167
Merge branch 'main' into haris/grpc-gateway-websockets
hariso 80c504d
rename
hariso 692a988
read loop
hariso e3a87a1
refactor + ping write loop
hariso 2072f7d
refactor
hariso a16ca05
fix pingWriteLoop
hariso 2ae9538
fixes
hariso eb7e986
comments
hariso b584fcf
remove ping-pong
hariso 600ef2f
Squashed commit of the following:
hariso b030581
Merge branch 'main' into haris/grpc-gateway-websockets
hariso b55a906
remove tests
hariso 47f6221
refactor
hariso b0ff958
fix concurrency
lovromazgon 7d4b410
comments
hariso 9fac27d
test
hariso b5da901
tests
hariso 2959aff
lint
hariso b8f1110
Update pkg/foundation/grpcutil/websocket.go
hariso 2601ad8
Update pkg/foundation/grpcutil/websocket.go
hariso File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,132 @@ | ||
| // Copyright © 2022 Meroxa, Inc. | ||
| // | ||
| // Licensed under the Apache License, Version 2.0 (the "License"); | ||
| // you may not use this file except in compliance with the License. | ||
| // You may obtain a copy of the License at | ||
| // | ||
| // http://www.apache.org/licenses/LICENSE-2.0 | ||
| // | ||
| // Unless required by applicable law or agreed to in writing, software | ||
| // distributed under the License is distributed on an "AS IS" BASIS, | ||
| // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| // See the License for the specific language governing permissions and | ||
| // limitations under the License. | ||
|
|
||
| package grpcutil | ||
|
|
||
| import ( | ||
| "bufio" | ||
| "context" | ||
| "io" | ||
| "net/http" | ||
|
|
||
| "github.com/conduitio/conduit/pkg/foundation/log" | ||
| "github.com/gorilla/websocket" | ||
| ) | ||
|
|
||
| type inMemoryResponseWriter struct { | ||
| io.Writer | ||
| header http.Header | ||
| closed chan bool | ||
| } | ||
|
|
||
| func newInMemoryResponseWriter(writer io.Writer) *inMemoryResponseWriter { | ||
| return &inMemoryResponseWriter{ | ||
| Writer: writer, | ||
| header: http.Header{}, | ||
| closed: make(chan bool, 1), | ||
| } | ||
| } | ||
|
|
||
| func (w *inMemoryResponseWriter) Write(b []byte) (int, error) { | ||
| return w.Writer.Write(b) | ||
| } | ||
| func (w *inMemoryResponseWriter) Header() http.Header { | ||
| return w.header | ||
| } | ||
| func (w *inMemoryResponseWriter) WriteHeader(int) { | ||
| // we don't have a use for the code | ||
| } | ||
| func (w *inMemoryResponseWriter) CloseNotify() <-chan bool { | ||
| return w.closed | ||
| } | ||
| func (w *inMemoryResponseWriter) Flush() {} | ||
|
|
||
| // wsProxy is a proxy around a http.Handler which | ||
| // redirects the response data from the http.Handler | ||
| // to a WebSocket connection. | ||
| type wsProxy struct { | ||
| handler http.Handler | ||
| logger log.CtxLogger | ||
| upgrader websocket.Upgrader | ||
| } | ||
|
|
||
| func newWebSocketProxy(handler http.Handler, logger log.CtxLogger) *wsProxy { | ||
| return &wsProxy{ | ||
| handler: handler, | ||
| logger: logger.WithComponent("grpcutil.websocket"), | ||
| upgrader: websocket.Upgrader{ | ||
| ReadBufferSize: 1024, | ||
| WriteBufferSize: 1024, | ||
| }, | ||
|
lovromazgon marked this conversation as resolved.
|
||
| } | ||
| } | ||
|
|
||
| func (p *wsProxy) ServeHTTP(w http.ResponseWriter, r *http.Request) { | ||
| if !websocket.IsWebSocketUpgrade(r) { | ||
| p.handler.ServeHTTP(w, r) | ||
| return | ||
| } | ||
| p.proxy(w, r) | ||
| } | ||
|
|
||
| func (p *wsProxy) proxy(w http.ResponseWriter, r *http.Request) { | ||
| ctx, cancelFn := context.WithCancel(r.Context()) | ||
| defer cancelFn() | ||
|
|
||
| // upgrade connection to WebSocket | ||
| conn, err := p.upgrader.Upgrade(w, r, http.Header{}) | ||
| if err != nil { | ||
| p.logger.Err(ctx, err).Msg("error upgrading websocket") | ||
| return | ||
| } | ||
| defer conn.Close() | ||
|
|
||
| // We use a pipe to read the data being written to the underlying http.Handler | ||
| // and then write it to the WebSocket connection. | ||
| responseR, responseW := io.Pipe() | ||
| response := newInMemoryResponseWriter(responseW) | ||
| go func() { | ||
| <-ctx.Done() | ||
| p.logger.Debug(ctx).Err(ctx.Err()).Msg("closing pipes") | ||
| responseW.CloseWithError(io.EOF) | ||
| response.closed <- true | ||
| }() | ||
|
|
||
| go func() { | ||
| defer cancelFn() | ||
| p.handler.ServeHTTP(response, r) | ||
| }() | ||
|
|
||
| scanner := bufio.NewScanner(responseR) | ||
|
|
||
| for scanner.Scan() { | ||
| if len(scanner.Bytes()) == 0 { | ||
| p.logger.Warn(ctx).Err(scanner.Err()).Msg("[write] empty scan") | ||
| continue | ||
| } | ||
|
|
||
| p.logger.Trace(ctx).Msgf("[write] scanned %v", scanner.Text()) | ||
| if err := conn.WriteMessage(websocket.TextMessage, scanner.Bytes()); err != nil { | ||
| p.logger.Warn(ctx).Err(err).Msg("[write] error writing websocket message") | ||
| return | ||
| } | ||
| } | ||
|
|
||
| if sErr := scanner.Err(); sErr != nil { | ||
| p.logger.Err(ctx, sErr).Msg("failed reading data from original response") | ||
| if err := conn.WriteMessage(websocket.TextMessage, []byte(sErr.Error())); err != nil { | ||
| p.logger.Warn(ctx).Err(err).Msg("[write] failed writing scanner error") | ||
| } | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,88 @@ | ||
| // Copyright © 2022 Meroxa, Inc. | ||
| // | ||
| // Licensed under the Apache License, Version 2.0 (the "License"); | ||
| // you may not use this file except in compliance with the License. | ||
| // You may obtain a copy of the License at | ||
| // | ||
| // http://www.apache.org/licenses/LICENSE-2.0 | ||
| // | ||
| // Unless required by applicable law or agreed to in writing, software | ||
| // distributed under the License is distributed on an "AS IS" BASIS, | ||
| // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| // See the License for the specific language governing permissions and | ||
| // limitations under the License. | ||
|
|
||
| package grpcutil | ||
|
|
||
| import ( | ||
| "context" | ||
| "io" | ||
| "net/http" | ||
| "net/http/httptest" | ||
| "strings" | ||
| "testing" | ||
|
|
||
| "github.com/conduitio/conduit/pkg/foundation/log" | ||
| "github.com/gorilla/websocket" | ||
| "github.com/matryer/is" | ||
| ) | ||
|
|
||
| type testHandler struct { | ||
| is *is.I | ||
| response string | ||
| } | ||
|
|
||
| func (h *testHandler) ServeHTTP(w http.ResponseWriter, _ *http.Request) { | ||
| _, err := w.Write([]byte(h.response)) | ||
| h.is.NoErr(err) | ||
| } | ||
|
|
||
| func TestWebSocket_NoUpgradeToWebSocket(t *testing.T) { | ||
|
hariso marked this conversation as resolved.
|
||
| is := is.New(t) | ||
| ctx, cancel := context.WithCancel(context.Background()) | ||
| defer cancel() | ||
|
|
||
| h := &testHandler{ | ||
| is: is, | ||
| response: "hi there", | ||
| } | ||
| s := httptest.NewServer(newWebSocketProxy(h, log.Nop())) | ||
| defer s.Close() | ||
|
|
||
| req, err := http.NewRequestWithContext(ctx, "GET", s.URL, nil) | ||
| is.NoErr(err) | ||
|
|
||
| resp, err := http.DefaultClient.Do(req) | ||
| is.NoErr(err) | ||
| is.True(resp.Body != nil) // expected response to have a body | ||
| defer resp.Body.Close() | ||
|
|
||
| bytes, err := io.ReadAll(resp.Body) | ||
| is.NoErr(err) | ||
| is.Equal(h.response, string(bytes)) | ||
| } | ||
|
|
||
| func TestWebSocket_UpgradeToWebSocket(t *testing.T) { | ||
| is := is.New(t) | ||
|
|
||
| h := &testHandler{ | ||
| is: is, | ||
| response: "hi there", | ||
| } | ||
| s := httptest.NewServer(newWebSocketProxy(h, log.Nop())) | ||
| defer s.Close() | ||
|
|
||
| // Convert http to ws | ||
| wsURL := "ws" + strings.TrimPrefix(s.URL, "http") | ||
|
|
||
| // Connect to the server | ||
| ws, resp, err := websocket.DefaultDialer.Dial(wsURL, nil) | ||
| is.NoErr(err) | ||
| defer ws.Close() | ||
| defer resp.Body.Close() | ||
|
|
||
| msgType, bytes, err := ws.ReadMessage() | ||
| is.NoErr(err) | ||
| is.Equal(h.response, string(bytes)) | ||
| is.Equal(websocket.TextMessage, msgType) | ||
| } | ||
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.