Messaging client library for use by Go implementation of EdgeX micro services. This project contains the abstract Message Bus interface and an implementation for Redis Pub/Sub, MQTT and NATS. These interface functions connect, publish, subscribe and disconnect to/from the Message Bus. For more information see the MessageBus documentation.
- Create new MessageClient
- Connect to the Message Bus
- Public messages to the Message Bus
- Subscribe to and receives messages from the Messsage Bus
- Disconnect from the Message Bus
- Make sure you have modules enabled, i.e. have an initialized go.mod file
- If your code is in your GOPATH then make sure
GO111MODULE=onis set - Run
go get github.com/edgexfoundry/go-mod-messaging/v3- This will add the go-mod-messaging to the go.mod file and download it into the module cache
This library is used by Go programs for interacting with the Message Bus (i.e. redis).
The Message Bus connection information as well as which implementation to use is stored in the service's toml configuration as:
MessageBus:
Protocol: redis
Host: localhost
Port: 6379
Type: redisIndividual client abstractions allow additional configuration properties which can be provided via configuration file:
MessageBus:
Protocol: tcp
Host: localhost
Port: 1883
Type: mqtt
Topic: events
Optional:
ClientId: MyClient
Username: MyUsername
...Or programmatically in the Optional field of the MessageBusConfig struct. For example,
types.MessageBusConfig{
Broker: types.HostInfo{Host: "example.com", Port: 9090, Protocol: "tcp"},
Optional: map[string]string{
"ClientId": "MyClientID",
"Username": "MyUser",
"Password": "MyPassword",
...
}}NOTE
For complete details on configuration options see the MessageBus documentation
The following code snippets demonstrate how a service uses this messaging module to create a connection, send messages, and receive messages.
This code snippet shows how to connect to the abstract message bus.
var messageBus messaging.MessageClient
var err error
messageBus, err = msgFactory.NewMessageClient(types.MessageBusConfig{
Broker: types.HostInfo{
Host: Configuration.MessageBus.Host,
Port: Configuration.MessageBus.Port,
Protocol: Configuration.MessageBus.Protocol,
},
Type: Configuration.MessageBus.Type,})
if err != nil {
LoggingClient.Error("failed to create messaging client: " + err.Error())
}
err = messsageBus.Connect()
if err != nil {
LoggingClient.Error("failed to connect to message bus: " + err.Error())
}This code snippet shows how to publish a message to the abstract message bus.
...
payload, err := json.Marshal(evt)
...
msgEnvelope := types.MessageEnvelope{
CorrelationID: evt.CorrelationId,
Payload: payload,
ContentType: clients.ContentJson,
}
err = messageBus.Publish(msgEnvelope, Configuration.MessageBus.Topic)This code snippet shows how to subscribe to the abstract message bus.
messageBus, err := factory.NewMessageClient(types.MessageBusConfig{
Broker: types.HostInfo{
Host: Configuration.MessageBus.Host,
Port: Configuration.MessageBus.Port,
Protocol: Configuration.MessageBus.Protocol,
},
Type: Configuration.MessageBus.Type,
})
if err != nil {
LoggingClient.Error("failed to create messaging client: " + err.Error())
return
}
if err := messageBus.Connect(); err != nil {
LoggingClient.Error("failed to connect to message bus: " + err.Error())
return
}
topics := []types.TopicChannel{
{
Topic: Configuration.MessageBus.Topic,
Messages: messages,
},
}
err = messageBus.Subscribe(topics, messageErrors)
if err != nil {
LoggingClient.Error("failed to subscribe for event messages: " + err.Error())
return
}This code snippet shows how to receive data on the message channel after you have subscribed to the bus.
...
for {
select {
case e := <-errors:
// handle errors
...
case msgEnvelope := <-messages:
LoggingClient.Info(fmt.Sprintf("Event received on message queue. Topic: %s, Correlation-id: %s ", Configuration.MessageBus.Topic, msgEnvelope.CorrelationID))
if msgEnvelope.ContentType != clients.ContentJson {
LoggingClient.Error(fmt.Sprintf("Incorrect content type for event message. Received: %s, Expected: %s", msgEnvelope.ContentType, clients.ContentJson))
continue
}
str := string(msgEnvelope.Payload)
event := parseEvent(str)
if event == nil {
continue
}
}
...
![dependabot[bot] avatar](https://avatars.githubusercontent.com/in/29110?v=4 "dependabot[bot]")
React
Typescript
Django
Laravel
Facebook
Microsoft
Google
Alibaba
D3
Tencent