forked from influxdata/kapacitor
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Much of this code was pulled from the guide published here: https://github.com/influxdata/kapacitor/blob/master/alert/HANDLERS.md . This also adds eclipse/paho.mqtt.golang as a dependency to Kapacitor. This client was chosen primarily because it is the same one that is used in Telegraf. It is actively developed, and is widely used by other popular projects in the community (e.g. Gobot).
- Loading branch information
1 parent
a16064c
commit c50c13c
Showing
72 changed files
with
9,938 additions
and
2 deletions.
There are no files selected for viewing
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
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
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,105 @@ | ||
| package mqtt | ||
|
|
||
| import ( | ||
| "errors" | ||
| "log" | ||
| "time" | ||
|
|
||
| pahomqtt "github.com/eclipse/paho.mqtt.golang" | ||
| ) | ||
|
|
||
| // Client describes an immutable MQTT client, designed to accommodate the | ||
| // incongruencies between real clients and mock clients. | ||
| type Client interface { | ||
| Connect() error | ||
| Disconnect() | ||
| Publish(string, byte, bool, string) error | ||
| } | ||
|
|
||
| // NewClient produces a disconnected MQTT client | ||
| var NewClient = func(c Config) Client { | ||
| return &PahoClient{ | ||
| host: c.Broker(), | ||
| port: c.Port, | ||
| username: c.Username, | ||
| password: c.Password, | ||
| clientID: c.ClientID, | ||
| } | ||
| } | ||
|
|
||
| type PahoClient struct { | ||
| host string | ||
| port uint16 | ||
|
|
||
| username string | ||
| password string | ||
|
|
||
| clientID string | ||
|
|
||
| client pahomqtt.Client | ||
| } | ||
|
|
||
| var _ Client = &PahoClient{} | ||
|
|
||
| // DefaultQuiesceTimeout is the duration the client will wait for outstanding | ||
| // messages to be published before forcing a disconnection | ||
| const DefaultQuiesceTimeout time.Duration = 250 * time.Millisecond | ||
|
|
||
| func (p *PahoClient) Connect() error { | ||
| log.Printf("Current config: %#v\n", p) | ||
| opts := pahomqtt.NewClientOptions() | ||
| opts.AddBroker(p.host) | ||
| opts.SetClientID(p.clientID) | ||
| opts.SetUsername(p.username) | ||
| opts.SetPassword(p.password) | ||
|
|
||
| // Using a clean session forces the broker to dispose of client session | ||
| // information after disconnecting. Retention of this is useful for | ||
| // constrained clients. Since Kapacitor is only publishing, it has no | ||
| // storage requirements and can reduce load on the broker by using a clean | ||
| // session. | ||
| opts.SetCleanSession(true) | ||
|
|
||
| p.client = pahomqtt.NewClient(opts) | ||
| token := p.client.Connect() | ||
| log.Printf("Current config: %#v\n", p) | ||
|
|
||
| token.Wait() // Tokens are futures | ||
|
|
||
| return token.Error() | ||
| } | ||
|
|
||
| func (p *PahoClient) Disconnect() { | ||
| if p.client != nil { | ||
| p.client.Disconnect(uint(DefaultQuiesceTimeout / time.Millisecond)) | ||
| } | ||
| } | ||
|
|
||
| var foo int = 1 | ||
|
|
||
| func (p *PahoClient) Publish(topic string, qos byte, retained bool, message string) error { | ||
| p.client.Publish(topic, qos, retained, message) | ||
| return nil | ||
| } | ||
|
|
||
| var _ Client = &MockClient{} | ||
|
|
||
| type MockClient struct { | ||
| connected bool | ||
| } | ||
|
|
||
| func (m *MockClient) Connect() error { | ||
| m.connected = true | ||
| return nil | ||
| } | ||
|
|
||
| func (m *MockClient) Disconnect() { | ||
| m.connected = false | ||
| } | ||
|
|
||
| func (m *MockClient) Publish(topic string, qos byte, retained bool, message string) error { | ||
| if !m.connected { | ||
| return errors.New("Publish() called before Connect()") | ||
| } | ||
| return nil | ||
| } |
Oops, something went wrong.