-
Notifications
You must be signed in to change notification settings - Fork 217
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* adding an example protocol based on the test protocol channel based protocol Signed-off-by: Scott Nichols <snichols@vmware.com> * Adding test for protocol. Signed-off-by: Scott Nichols <snichols@vmware.com> * Promote the test protocol to a real protocol. Signed-off-by: Scott Nichols <snichols@vmware.com> * No need to chan chan all over. Signed-off-by: Scott Nichols <snichols@vmware.com>
- Loading branch information
Showing
11 changed files
with
337 additions
and
97 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,50 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "context" | ||
| cloudevents "github.com/cloudevents/sdk-go/v2" | ||
| "github.com/cloudevents/sdk-go/v2/protocol/gochan" | ||
| "log" | ||
| "time" | ||
| ) | ||
|
|
||
| func main() { | ||
| c, err := cloudevents.NewClient(gochan.New(), cloudevents.WithTimeNow(), cloudevents.WithUUIDs()) | ||
| if err != nil { | ||
| log.Fatalf("failed to create client: %v", err) | ||
| } | ||
|
|
||
| ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(time.Millisecond*50)) // wait | ||
|
|
||
| // Start the receiver | ||
| go func() { | ||
| if err := c.StartReceiver(ctx, func(ctx context.Context, event cloudevents.Event) { | ||
| log.Printf("[receiver] %s", event) | ||
| }); err != nil && err.Error() != "context deadline exceeded" { | ||
| log.Fatalf("[receiver] start receiver returned an error: %s", err) | ||
| } | ||
| log.Println("[receiver] stopped") | ||
| }() | ||
|
|
||
| // Start sending the events | ||
| for i := 0; i < 10; i++ { | ||
| e := cloudevents.NewEvent() | ||
| e.SetType("com.cloudevents.sample.sent") | ||
| e.SetSource("https://github.com/cloudevents/sdk-go/v2/cmd/samples/gochan") | ||
| _ = e.SetData(cloudevents.ApplicationJSON, map[string]interface{}{ | ||
| "id": i, | ||
| "message": "Hello, World!", | ||
| }) | ||
|
|
||
| err := c.Send(ctx, e) | ||
| if err != nil { | ||
| log.Printf("[sender] failed to send: %v", err) | ||
| } else { | ||
| log.Printf("[sender] sent: %d", i) | ||
| } | ||
| } | ||
| // Wait for the timeout. | ||
| <-ctx.Done() | ||
| cancel() | ||
| log.Println("[sender] stopped") | ||
| } |
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,4 @@ | ||
| /* | ||
| Package channel implements the CloudEvent transport implementation using go chan. | ||
| */ | ||
| package gochan |
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,35 @@ | ||
| package gochan | ||
|
|
||
| import ( | ||
| "context" | ||
| "github.com/cloudevents/sdk-go/v2/binding" | ||
| "github.com/cloudevents/sdk-go/v2/protocol" | ||
| ) | ||
|
|
||
| const ( | ||
| defaultChanDepth = 20 | ||
| ) | ||
|
|
||
| // SendReceiver is a reference implementation for using the CloudEvents binding | ||
| // integration. | ||
| type SendReceiver struct { | ||
| sender protocol.Sender | ||
| receiver protocol.Receiver | ||
| } | ||
|
|
||
| func New() *SendReceiver { | ||
| ch := make(chan binding.Message, defaultChanDepth) | ||
|
|
||
| return &SendReceiver{ | ||
| sender: Sender(ch), | ||
| receiver: Receiver(ch), | ||
| } | ||
| } | ||
|
|
||
| func (s *SendReceiver) Send(ctx context.Context, in binding.Message) (err error) { | ||
| return s.sender.Send(ctx, in) | ||
| } | ||
|
|
||
| func (r *SendReceiver) Receive(ctx context.Context) (binding.Message, error) { | ||
| return r.receiver.Receive(ctx) | ||
| } |
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,140 @@ | ||
| package gochan | ||
|
|
||
| import ( | ||
| "context" | ||
| "github.com/cloudevents/sdk-go/v2/binding" | ||
| "github.com/cloudevents/sdk-go/v2/event" | ||
| "github.com/google/go-cmp/cmp" | ||
| "github.com/stretchr/testify/assert" | ||
| "testing" | ||
| "time" | ||
| ) | ||
|
|
||
| func TestNew(t *testing.T) { | ||
| got := New() | ||
| assert.NotNil(t, got) | ||
| } | ||
|
|
||
| func protocols(t *testing.T) []*SendReceiver { | ||
| return []*SendReceiver{New()} | ||
| } | ||
|
|
||
| func TestSend(t *testing.T) { | ||
| testCases := map[string]struct { | ||
| ctx context.Context | ||
| msg binding.Message | ||
| wantErr string | ||
| }{ | ||
| "nil context": { | ||
| wantErr: "nil Context", | ||
| }, | ||
| "nil message": { | ||
| ctx: context.TODO(), | ||
| wantErr: "nil Message", | ||
| }, | ||
| } | ||
| for n, tc := range testCases { | ||
| for _, p := range protocols(t) { | ||
| t.Run(n, func(t *testing.T) { | ||
| err := p.Send(tc.ctx, tc.msg) | ||
| if tc.wantErr != "" { | ||
| if err == nil || err.Error() != tc.wantErr { | ||
| t.Fatalf("Expected error '%s'. Actual '%v'", tc.wantErr, err) | ||
| } | ||
| } else if err != nil { | ||
| t.Fatalf("Unexpected error: %v", err) | ||
| } | ||
| }) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func TestReceive(t *testing.T) { | ||
| testCases := map[string]struct { | ||
| ctx context.Context | ||
| want binding.Message | ||
| wantErr string | ||
| }{ | ||
| "nil context": { | ||
| wantErr: "nil Context", | ||
| }, | ||
| "timeout": { | ||
| ctx: context.TODO(), | ||
| wantErr: "context deadline exceeded", | ||
| }, | ||
| } | ||
| for n, tc := range testCases { | ||
| for _, p := range protocols(t) { | ||
| t.Run(n, func(t *testing.T) { | ||
| ReceiveTest(t, p, tc.ctx, tc.want, tc.wantErr) | ||
| }) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func TestSendReceive(t *testing.T) { | ||
| testCases := map[string]struct { | ||
| sendErr string | ||
| want binding.Message | ||
| receiveErr string | ||
| }{ | ||
| "nil": { | ||
| sendErr: "nil Message", | ||
| receiveErr: "context deadline exceeded", | ||
| }, | ||
| "empty event": { | ||
| want: func() binding.Message { | ||
| e := event.New() | ||
| return binding.ToMessage(&e) | ||
| }(), | ||
| }, | ||
| "min event": { | ||
| want: func() binding.Message { | ||
| e := event.New() | ||
| e.SetSource("unittest/") | ||
| e.SetType("unit.test") | ||
| e.SetID("unit-test") | ||
| return binding.ToMessage(&e) | ||
| }(), | ||
| }, | ||
| } | ||
| for n, tc := range testCases { | ||
| for _, p := range protocols(t) { | ||
| t.Run(n, func(t *testing.T) { | ||
| go func() { | ||
| wantErr := tc.sendErr | ||
| err := p.Send(context.Background(), tc.want) | ||
| if wantErr != "" { | ||
| if err == nil || err.Error() != wantErr { | ||
| t.Fatalf("Expected error '%s'. Actual '%v'", wantErr, err) | ||
| } | ||
| } else if err != nil { | ||
| t.Fatalf("Unexpected error: %v", err) | ||
| } | ||
| }() | ||
| ReceiveTest(t, p, context.Background(), tc.want, tc.receiveErr) | ||
| }) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func ReceiveTest(t *testing.T, p *SendReceiver, ctx context.Context, want binding.Message, wantErr string) { | ||
| if ctx != nil { | ||
| var done context.CancelFunc | ||
| ctx, done = context.WithTimeout(ctx, time.Millisecond*10) | ||
| defer done() | ||
| } | ||
|
|
||
| got, err := p.Receive(ctx) | ||
| if wantErr != "" { | ||
| if err == nil || err.Error() != wantErr { | ||
| t.Fatalf("Expected error '%s'. Actual '%v'", wantErr, err) | ||
| } | ||
| } else if err != nil { | ||
| t.Fatalf("Unexpected error: %v", err) | ||
| } | ||
|
|
||
| if diff := cmp.Diff(want, got); diff != "" { | ||
| t.Errorf("unexpected diff (-want, +got) = %v", diff) | ||
| } | ||
| } |
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,31 @@ | ||
| package gochan | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "github.com/cloudevents/sdk-go/v2/protocol" | ||
| "io" | ||
|
|
||
| "github.com/cloudevents/sdk-go/v2/binding" | ||
| ) | ||
|
|
||
| // Receiver implements Receiver by receiving Messages from a channel. | ||
| type Receiver <-chan binding.Message | ||
|
|
||
| func (r Receiver) Receive(ctx context.Context) (binding.Message, error) { | ||
| if ctx == nil { | ||
| return nil, fmt.Errorf("nil Context") | ||
| } | ||
|
|
||
| select { | ||
| case <-ctx.Done(): | ||
| return nil, ctx.Err() | ||
| case m, ok := <-r: | ||
| if !ok { | ||
| return nil, io.EOF | ||
| } | ||
| return m, nil | ||
| } | ||
| } | ||
|
|
||
| var _ protocol.Receiver = (*Receiver)(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
Oops, something went wrong.