-
Notifications
You must be signed in to change notification settings - Fork 83
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(sdk): Add background publisher to MessageBus
Creates a channel of provided capacity and attaches to the sdk instance. This channel is then passed to the trigger so that MessageEnvelopes can be pulled off of it and dropped on the queue. Publisher handles formatting the passed message so there is no need to couple using types from go-mod-messaging. Closes: #462 Signed-off-by: Alex Ullrich <[email protected]>
- Loading branch information
Alex Ullrich
committed
Sep 4, 2020
1 parent
337bfa7
commit aa9f94a
Showing
9 changed files
with
261 additions
and
12 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,46 @@ | ||
| // | ||
| // Copyright (c) 2020 Technotects | ||
| // | ||
| // 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 appsdk | ||
|
|
||
| import "github.com/edgexfoundry/go-mod-messaging/pkg/types" | ||
|
|
||
| // BackgroundPublisher provides an interface to send messages from background processes | ||
| // through the service's configured MessageBus output | ||
| type BackgroundPublisher interface { | ||
| // Publish provided message through the configured MessageBus output | ||
| Publish(payload []byte, correlationID string, contentType string) | ||
| } | ||
|
|
||
| type backgroundPublisher struct { | ||
| output chan<- types.MessageEnvelope | ||
| } | ||
|
|
||
| // Publish provided message through the configured MessageBus output | ||
| func (pub *backgroundPublisher) Publish(payload []byte, correlationID string, contentType string) { | ||
| outputEnvelope := types.MessageEnvelope{ | ||
| CorrelationID: correlationID, | ||
| Payload: payload, | ||
| ContentType: contentType, | ||
| } | ||
|
|
||
| pub.output <- outputEnvelope | ||
| } | ||
|
|
||
| func newBackgroundPublisher(capacity int) (<-chan types.MessageEnvelope, BackgroundPublisher) { | ||
| backgroundChannel := make(chan types.MessageEnvelope, capacity) | ||
| return backgroundChannel, &backgroundPublisher{output: backgroundChannel} | ||
| } |
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,48 @@ | ||
| // | ||
| // Copyright (c) 2020 Technotects | ||
| // | ||
| // 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 appsdk | ||
|
|
||
| import ( | ||
| "github.com/stretchr/testify/assert" | ||
| "testing" | ||
| "time" | ||
| ) | ||
|
|
||
| func TestNewBackgroundPublisherAndPublish(t *testing.T) { | ||
| background, pub := newBackgroundPublisher(1) | ||
|
|
||
| payload := []byte("something") | ||
| correlationId := "id" | ||
| contentType := "type" | ||
|
|
||
| pub.Publish(payload, correlationId, contentType) | ||
|
|
||
| waiting := true | ||
|
|
||
| for waiting { | ||
| select { | ||
| case msgs := <-background: | ||
| assert.Equal(t, correlationId, msgs.CorrelationID) | ||
| assert.Equal(t, contentType, msgs.ContentType) | ||
| assert.Equal(t, payload, msgs.Payload) | ||
| waiting = false | ||
| case <-time.After(1 * time.Second): | ||
| assert.Fail(t, "message timed out, background channel likely not configured correctly") | ||
| waiting = false | ||
| } | ||
| } | ||
| } |
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,34 @@ | ||
| // | ||
| // Copyright (c) 2020 Technotects | ||
| // | ||
| // 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 http | ||
|
|
||
| import ( | ||
| "github.com/edgexfoundry/go-mod-messaging/pkg/types" | ||
| "github.com/stretchr/testify/assert" | ||
| "testing" | ||
| ) | ||
|
|
||
| func TestTriggerInitializeWitBackgroundChannel(t *testing.T) { | ||
| background := make(chan types.MessageEnvelope) | ||
| trigger := Trigger{} | ||
|
|
||
| deferred, err := trigger.Initialize(nil, nil, background) | ||
|
|
||
| assert.Nil(t, deferred) | ||
| assert.NotNil(t, err) | ||
| assert.Equal(t, "background publishing not supported for services using HTTP trigger", err.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
Oops, something went wrong.