This repository has been archived by the owner on Aug 21, 2021. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 56
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #155 from NoteGio/feature/ws
Feature/ws
- Loading branch information
Showing
98 changed files
with
9,783 additions
and
90 deletions.
There are no files selected for viewing
This file contains 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 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,9 @@ | ||
| FROM corebuild | ||
|
|
||
| FROM scratch | ||
|
|
||
| COPY --from=corebuild /go/src/github.com/notegio/openrelay/bin/websockets /websockets | ||
|
|
||
| COPY --from=corebuild /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/ca-certificates.crt | ||
|
|
||
| CMD ["/websockets", "redis:6379", "topic://released", "postgres://postgres@postgres", "/run/secrets/postgress_password"] |
This file contains 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 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,126 @@ | ||
| package ws | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "github.com/notegio/openrelay/channels" | ||
| "github.com/notegio/openrelay/types" | ||
| "github.com/notegio/openrelay/pool" | ||
| "github.com/gorilla/websocket" | ||
| "github.com/jinzhu/gorm" | ||
| "net/http" | ||
| "log" | ||
| ) | ||
|
|
||
| var upgrader = websocket.Upgrader{ | ||
| ReadBufferSize: 1024, | ||
| WriteBufferSize: 1024, | ||
| } | ||
|
|
||
| type websocketDelivery struct { | ||
| payload string | ||
| } | ||
|
|
||
| func (delivery *websocketDelivery) Payload() string { | ||
| return delivery.payload | ||
| } | ||
|
|
||
| func (delivery *websocketDelivery) Ack() bool { | ||
| // websocketDeliveris have no ack, reject, or return, so these are no-ops | ||
| return true | ||
| } | ||
| func (delivery *websocketDelivery) Reject() bool { | ||
| return true | ||
| } | ||
| func (delivery *websocketDelivery) Return() bool { | ||
| return true | ||
| } | ||
|
|
||
| type WebsocketChannel struct { | ||
| open bool | ||
| conn *websocket.Conn | ||
| payloads chan []byte | ||
| consumers []channels.Consumer | ||
| Filter string | ||
| quit chan struct{} | ||
| cleanup func(channels.Publisher) | ||
| } | ||
|
|
||
| func (pub *WebsocketChannel) Publish(payload string) bool { | ||
| select { | ||
| case pub.payloads <- []byte(payload): | ||
| return true | ||
| default: | ||
| return false | ||
| } | ||
| } | ||
|
|
||
| func (consumerChannel *WebsocketChannel) AddConsumer(consumer channels.Consumer) bool { | ||
| consumerChannel.consumers = append(consumerChannel.consumers, consumer) | ||
| return true | ||
| } | ||
| func (consumerChannel *WebsocketChannel) StartConsuming() bool { | ||
| go func () { | ||
| defer consumerChannel.cleanup(consumerChannel) | ||
| for { | ||
| select { | ||
| case _ = <-consumerChannel.quit: | ||
| return | ||
| default: | ||
| } | ||
| _, p, err := consumerChannel.conn.ReadMessage() | ||
| if err != nil { | ||
| log.Println(err) | ||
| return | ||
| } | ||
| for _, consumer := range consumerChannel.consumers { | ||
| consumer.Consume(&websocketDelivery{string(p)}) | ||
| } | ||
| } | ||
| }() | ||
| return true | ||
| } | ||
|
|
||
| func (consumerChannel *WebsocketChannel) StopConsuming() bool { | ||
| consumerChannel.quit <- struct{}{} | ||
| return true | ||
| } | ||
| func (consumerChannel *WebsocketChannel) ReturnAllUnacked() int { | ||
| return 0 | ||
| } | ||
| func (consumerChannel *WebsocketChannel) PurgeRejected() int { | ||
| return 0 | ||
| } | ||
| func (consumerChannel *WebsocketChannel) Publisher() channels.Publisher { | ||
| return consumerChannel | ||
| } | ||
|
|
||
| func GetChannels(port uint, db *gorm.DB, cleanup func(channels.Publisher)) (<-chan *WebsocketChannel, func() (error)) { | ||
| outChan := make(chan *WebsocketChannel) | ||
| handler := pool.PoolDecorator(db, func (w http.ResponseWriter, r *http.Request, p types.Pool) { | ||
| conn, err := upgrader.Upgrade(w, r, nil) | ||
| if err != nil { | ||
| log.Println(err) | ||
| return | ||
| } | ||
| wsChannel := &WebsocketChannel{true, conn, make(chan []byte), []channels.Consumer{}, p.QueryString(), make(chan struct{}), cleanup} | ||
| outChan <- wsChannel | ||
| for payload := range wsChannel.payloads { | ||
| if err := conn.WriteMessage(websocket.BinaryMessage, payload); err != nil { | ||
| log.Println(err) | ||
| return | ||
| } | ||
| } | ||
| }) | ||
|
|
||
| mux := http.NewServeMux() | ||
| mux.HandleFunc("/", handler) | ||
| srv := &http.Server{ | ||
| Addr: fmt.Sprintf(":%v", port), | ||
| Handler: mux, | ||
| } | ||
| go func() { | ||
| log.Printf("%v", srv.ListenAndServe()) | ||
| }() | ||
| return outChan, func() (error) { return srv.Shutdown(context.Background()) } | ||
| } |
This file contains 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,64 @@ | ||
| package ws_test | ||
|
|
||
| import ( | ||
| "context" | ||
| "github.com/notegio/openrelay/channels" | ||
| "github.com/notegio/openrelay/channels/ws" | ||
| "github.com/gorilla/websocket" | ||
| "testing" | ||
| "time" | ||
| // "log" | ||
| ) | ||
|
|
||
| type TestConsumer struct { | ||
| channel *ws.WebsocketChannel | ||
| } | ||
|
|
||
| func (consumer *TestConsumer) Consume(delivery channels.Delivery) { | ||
| if delivery.Payload() == "quit" { | ||
| consumer.channel.StopConsuming() | ||
| delivery.Ack() | ||
| } | ||
| consumer.channel.Publish(delivery.Payload()) | ||
| delivery.Ack() | ||
| } | ||
|
|
||
| func TestGetChannels(t *testing.T) { | ||
| clean := false | ||
| channels, quit := ws.GetChannels(4321, nil, func(channels.Publisher) { clean = true }) | ||
| go func() { | ||
| for channel := range channels { | ||
| channel.AddConsumer(&TestConsumer{channel}) | ||
| channel.StartConsuming() | ||
| } | ||
| }() | ||
| ctx, cancel := context.WithCancel(context.Background()) | ||
| c, resp, err := websocket.DefaultDialer.DialContext(ctx, "ws://localhost:4321/v2/", nil) | ||
| defer cancel() | ||
| if err != nil { | ||
| content := []byte{} | ||
| resp.Body.Read(content[:]) | ||
| t.Fatalf("%v - (%v) %v", err.Error(), resp.StatusCode, content) | ||
| } | ||
| if err := c.WriteMessage(websocket.BinaryMessage, []byte("ping")); err != nil { | ||
| t.Errorf(err.Error()) | ||
| } | ||
| mtype, p, err := c.ReadMessage() | ||
| if err != nil { | ||
| t.Errorf(err.Error()) | ||
| } | ||
| if mtype != websocket.BinaryMessage { | ||
| t.Errorf("Unexpected message type %v", mtype) | ||
| } | ||
| if string(p) != "ping" { | ||
| t.Errorf("Unexpected message: %v", string(p)) | ||
| } | ||
| c.Close() | ||
| if err := quit(); err != nil { | ||
| t.Errorf(err.Error()) | ||
| } | ||
| time.Sleep(50 * time.Millisecond) | ||
| if clean != true { | ||
| t.Errorf("Should have cleaned up") | ||
| } | ||
| } |
This file contains 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 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 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 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
Oops, something went wrong.