This repository has been archived by the owner on Apr 12, 2024. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 19
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[fix] Add missing STOMP over TCP option for Plank (#42)
* [fix] Add missing STOMP over TCP option for Plank Signed-off-by: Josh Kim <kjosh@vmware.com> * [fix] incorrect outbound send call for incoming msg Signed-off-by: Josh Kim <kjosh@vmware.com> * (new) broker_sample example talking to another Plank via TCP Signed-off-by: Josh Kim <kjosh@vmware.com> * fix: handle server start errors properly Signed-off-by: Josh Kim <kjosh@vmware.com> * test: increase test coverage Signed-off-by: Josh Kim <kjosh@vmware.com> * change: simplify sample code using CastPayloadToType Signed-off-by: Josh Kim <kjosh@vmware.com>
- Loading branch information
Showing
15 changed files
with
425 additions
and
36 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,72 @@ | ||
| package plank | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "github.com/vmware/transport-go/bridge" | ||
| "github.com/vmware/transport-go/bus" | ||
| "os" | ||
| "syscall" | ||
| "time" | ||
| ) | ||
|
|
||
| // ListenViaStomp listens to another Plank instance via TCP | ||
| func ListenViaStomp(c chan os.Signal) { | ||
| brokerConn, err := connectToBroker() | ||
| if err != nil { | ||
| fmt.Println(fmt.Errorf("broker connection failed: %w", err)) | ||
| os.Exit(1) | ||
| } | ||
|
|
||
| // subscribe to topic simple-stream which keeps sending a random word at an interval | ||
| sub, err := brokerConn.Subscribe("/topic/simple-stream") | ||
| if err != nil { | ||
| fmt.Println(fmt.Errorf("subscription failed: %w", err)) | ||
| os.Exit(1) | ||
| } | ||
|
|
||
| // let's disconnect after 10 seconds for the sake of an example | ||
| disconnectInTime := 10 * time.Second | ||
| tenSecTimer := time.NewTimer(disconnectInTime) | ||
|
|
||
| go func() { | ||
| for { | ||
| select { | ||
| case msg := <-sub.GetMsgChannel(): | ||
| // extract payload (of string type) from the Message object | ||
| var payload string | ||
| if err := msg.CastPayloadToType(&payload); err != nil { | ||
| fmt.Printf("failed to cast payload: %s\n", err.Error()) | ||
| continue | ||
| } | ||
| fmt.Printf("msg body: %s\n", payload) | ||
| break | ||
| case <-tenSecTimer.C: | ||
| fmt.Println(disconnectInTime.String() + " elapsed. Disconnecting") | ||
| tenSecTimer.Stop() | ||
| c <- syscall.SIGINT | ||
| break | ||
| } | ||
| } | ||
| }() | ||
|
|
||
| <-c | ||
| } | ||
|
|
||
| // connectToBroker wraps the connection logic to the broker and returns the bridge connection object and an error | ||
| func connectToBroker() (bridge.Connection, error) { | ||
| b := bus.GetBus() | ||
| brokerConfig := &bridge.BrokerConnectorConfig{ | ||
| Username: "guest", | ||
| Password: "guest", | ||
| ServerAddr: ":61613", | ||
| HeartBeatOut: 30 * time.Second, | ||
| } | ||
|
|
||
| brokerConn, err := b.ConnectBroker(brokerConfig) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| fmt.Println("broker connected. broker ID:", brokerConn.GetId()) | ||
| return brokerConn, nil | ||
| } |
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,36 @@ | ||
| package server | ||
|
|
||
| import "fmt" | ||
|
|
||
| var ( | ||
| errServerInit = &baseError{message: "Server initialization failed"} | ||
| errHttp = &baseError{message: "HTTP error"} | ||
| errInternal = &baseError{message: "Internal error"} | ||
| errUndefined = &baseError{message: "Undefined error"} | ||
| ) | ||
|
|
||
| type baseError struct { | ||
| wrappedErr error | ||
| baseErr *baseError | ||
| message string | ||
| } | ||
|
|
||
| func (e baseError) Is(err error) bool { | ||
| return e.baseErr == err | ||
| } | ||
|
|
||
| func (e baseError) Error() string { | ||
| return fmt.Sprintf("[plank] Error: %s: %s\n", e.baseErr.message, e.wrappedErr.Error()) | ||
| } | ||
|
|
||
| func wrapError(baseType error, err error) error { | ||
| switch baseType { | ||
| case errServerInit: | ||
| return &baseError{baseErr: errServerInit, wrappedErr: err} | ||
| case errInternal: | ||
| return &baseError{baseErr: errInternal, wrappedErr: err} | ||
| case errHttp: | ||
| return &baseError{baseErr: errHttp, wrappedErr: err} | ||
| } | ||
| return &baseError{baseErr: errUndefined, wrappedErr: err} | ||
| } |
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,47 @@ | ||
| package server | ||
|
|
||
| import ( | ||
| "errors" | ||
| "github.com/stretchr/testify/assert" | ||
| "testing" | ||
| ) | ||
|
|
||
| func TestBaseError_Is_errServerInit(t *testing.T) { | ||
| e := wrapError(errServerInit, errors.New("some init fail")) | ||
| assert.True(t, errors.Is(e, errServerInit)) | ||
| } | ||
|
|
||
| func TestBaseError_Error_errServerInit(t *testing.T) { | ||
| e := wrapError(errServerInit, errors.New("some init fail")) | ||
| assert.EqualValues(t, "[plank] Error: Server initialization failed: some init fail\n", e.Error()) | ||
| } | ||
|
|
||
| func TestBaseError_Is_errInternal(t *testing.T) { | ||
| e := wrapError(errInternal, errors.New("internal server error")) | ||
| assert.True(t, errors.Is(e, errInternal)) | ||
| } | ||
|
|
||
| func TestBaseError_Error_errInternal(t *testing.T) { | ||
| e := wrapError(errInternal, errors.New("internal server error")) | ||
| assert.EqualValues(t, "[plank] Error: Internal error: internal server error\n", e.Error()) | ||
| } | ||
|
|
||
| func TestBaseError_Is_errHttp(t *testing.T) { | ||
| e := wrapError(errHttp, errors.New("404")) | ||
| assert.True(t, errors.Is(e, errHttp)) | ||
| } | ||
|
|
||
| func TestBaseError_Error_errHttp(t *testing.T) { | ||
| e := wrapError(errHttp, errors.New("404")) | ||
| assert.EqualValues(t, "[plank] Error: HTTP error: 404\n", e.Error()) | ||
| } | ||
|
|
||
| func TestBaseError_Is_undefined(t *testing.T) { | ||
| e := wrapError(errors.New("some random stuff"), errors.New("?")) | ||
| assert.True(t, errors.Is(e, errUndefined)) | ||
| } | ||
|
|
||
| func TestBaseError_Error_undefined(t *testing.T) { | ||
| e := wrapError(errors.New("some random stuff"), errors.New("?")) | ||
| assert.EqualValues(t, "[plank] Error: Undefined error: ?\n", e.Error()) | ||
| } |
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.