robocars/robocar-steering
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

build: upgrade to go 1.17 and dependencies

Browse Source
This commit is contained in:
Cyrille Nofficial
2021-09-01 22:16:57 +02:00
parent ff49d76881
commit 210406c6db
461 changed files with 79514 additions and 99238 deletions
Download Patch File Download Diff File Expand all files Collapse all files

91
vendor/github.com/eclipse/paho.mqtt.golang/LICENSE generated vendored
View File

@ -1,87 +1,20 @@
Eclipse Public License - v 1.0
This project is dual licensed under the Eclipse Public License 1.0 and the
Eclipse Distribution License 1.0 as described in the epl-v10 and edl-v10 files.
THE ACCOMPANYING PROGRAM IS PROVIDED UNDER THE TERMS OF THIS ECLIPSE PUBLIC LICENSE ("AGREEMENT"). ANY USE, REPRODUCTION OR DISTRIBUTION OF THE PROGRAM CONSTITUTES RECIPIENT'S ACCEPTANCE OF THIS AGREEMENT.
The EDL is copied below in order to pass the pkg.go.dev license check (https://pkg.go.dev/license-policy).
1. DEFINITIONS
****
Eclipse Distribution License - v 1.0
"Contribution" means:
Copyright (c) 2007, Eclipse Foundation, Inc. and its licensors.
a) in the case of the initial Contributor, the initial code and documentation distributed under this Agreement, and
All rights reserved.
b) in the case of each subsequent Contributor:
Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met:
i) changes to the Program, and
Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer.
Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution.
Neither the name of the Eclipse Foundation, Inc. nor the names of its contributors may be used to endorse or promote products derived from this software without specific prior written permission.
ii) additions to the Program;
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
where such changes and/or additions to the Program originate from and are distributed by that particular Contributor. A Contribution 'originates' from a Contributor if it was added to the Program by such Contributor itself or anyone acting on such Contributor's behalf. Contributions do not include additions to the Program which: (i) are separate modules of software distributed in conjunction with the Program under their own license agreement, and (ii) are not derivative works of the Program.
"Contributor" means any person or entity that distributes the Program.
"Licensed Patents" mean patent claims licensable by a Contributor which are necessarily infringed by the use or sale of its Contribution alone or when combined with the Program.
"Program" means the Contributions distributed in accordance with this Agreement.
"Recipient" means anyone who receives the Program under this Agreement, including all Contributors.
2. GRANT OF RIGHTS
a) Subject to the terms of this Agreement, each Contributor hereby grants Recipient a non-exclusive, worldwide, royalty-free copyright license to reproduce, prepare derivative works of, publicly display, publicly perform, distribute and sublicense the Contribution of such Contributor, if any, and such derivative works, in source code and object code form.
b) Subject to the terms of this Agreement, each Contributor hereby grants Recipient a non-exclusive, worldwide, royalty-free patent license under Licensed Patents to make, use, sell, offer to sell, import and otherwise transfer the Contribution of such Contributor, if any, in source code and object code form. This patent license shall apply to the combination of the Contribution and the Program if, at the time the Contribution is added by the Contributor, such addition of the Contribution causes such combination to be covered by the Licensed Patents. The patent license shall not apply to any other combinations which include the Contribution. No hardware per se is licensed hereunder.
c) Recipient understands that although each Contributor grants the licenses to its Contributions set forth herein, no assurances are provided by any Contributor that the Program does not infringe the patent or other intellectual property rights of any other entity. Each Contributor disclaims any liability to Recipient for claims brought by any other entity based on infringement of intellectual property rights or otherwise. As a condition to exercising the rights and licenses granted hereunder, each Recipient hereby assumes sole responsibility to secure any other intellectual property rights needed, if any. For example, if a third party patent license is required to allow Recipient to distribute the Program, it is Recipient's responsibility to acquire that license before distributing the Program.
d) Each Contributor represents that to its knowledge it has sufficient copyright rights in its Contribution, if any, to grant the copyright license set forth in this Agreement.
3. REQUIREMENTS
A Contributor may choose to distribute the Program in object code form under its own license agreement, provided that:
a) it complies with the terms and conditions of this Agreement; and
b) its license agreement:
i) effectively disclaims on behalf of all Contributors all warranties and conditions, express and implied, including warranties or conditions of title and non-infringement, and implied warranties or conditions of merchantability and fitness for a particular purpose;
ii) effectively excludes on behalf of all Contributors all liability for damages, including direct, indirect, special, incidental and consequential damages, such as lost profits;
iii) states that any provisions which differ from this Agreement are offered by that Contributor alone and not by any other party; and
iv) states that source code for the Program is available from such Contributor, and informs licensees how to obtain it in a reasonable manner on or through a medium customarily used for software exchange.
When the Program is made available in source code form:
a) it must be made available under this Agreement; and
b) a copy of this Agreement must be included with each copy of the Program.
Contributors may not remove or alter any copyright notices contained within the Program.
Each Contributor must identify itself as the originator of its Contribution, if any, in a manner that reasonably allows subsequent Recipients to identify the originator of the Contribution.
4. COMMERCIAL DISTRIBUTION
Commercial distributors of software may accept certain responsibilities with respect to end users, business partners and the like. While this license is intended to facilitate the commercial use of the Program, the Contributor who includes the Program in a commercial product offering should do so in a manner which does not create potential liability for other Contributors. Therefore, if a Contributor includes the Program in a commercial product offering, such Contributor ("Commercial Contributor") hereby agrees to defend and indemnify every other Contributor ("Indemnified Contributor") against any losses, damages and costs (collectively "Losses") arising from claims, lawsuits and other legal actions brought by a third party against the Indemnified Contributor to the extent caused by the acts or omissions of such Commercial Contributor in connection with its distribution of the Program in a commercial product offering. The obligations in this section do not apply to any claims or Losses relating to any actual or alleged intellectual property infringement. In order to qualify, an Indemnified Contributor must: a) promptly notify the Commercial Contributor in writing of such claim, and b) allow the Commercial Contributor to control, and cooperate with the Commercial Contributor in, the defense and any related settlement negotiations. The Indemnified Contributor may participate in any such claim at its own expense.
For example, a Contributor might include the Program in a commercial product offering, Product X. That Contributor is then a Commercial Contributor. If that Commercial Contributor then makes performance claims, or offers warranties related to Product X, those performance claims and warranties are such Commercial Contributor's responsibility alone. Under this section, the Commercial Contributor would have to defend claims against the other Contributors related to those performance claims and warranties, and if a court requires any other Contributor to pay any damages as a result, the Commercial Contributor must pay those damages.
5. NO WARRANTY
EXCEPT AS EXPRESSLY SET FORTH IN THIS AGREEMENT, THE PROGRAM IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, EITHER EXPRESS OR IMPLIED INCLUDING, WITHOUT LIMITATION, ANY WARRANTIES OR CONDITIONS OF TITLE, NON-INFRINGEMENT, MERCHANTABILITY OR FITNESS FOR A PARTICULAR PURPOSE. Each Recipient is solely responsible for determining the appropriateness of using and distributing the Program and assumes all risks associated with its exercise of rights under this Agreement , including but not limited to the risks and costs of program errors, compliance with applicable laws, damage to or loss of data, programs or equipment, and unavailability or interruption of operations.
6. DISCLAIMER OF LIABILITY
EXCEPT AS EXPRESSLY SET FORTH IN THIS AGREEMENT, NEITHER RECIPIENT NOR ANY CONTRIBUTORS SHALL HAVE ANY LIABILITY FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING WITHOUT LIMITATION LOST PROFITS), HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OR DISTRIBUTION OF THE PROGRAM OR THE EXERCISE OF ANY RIGHTS GRANTED HEREUNDER, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGES.
7. GENERAL
If any provision of this Agreement is invalid or unenforceable under applicable law, it shall not affect the validity or enforceability of the remainder of the terms of this Agreement, and without further action by the parties hereto, such provision shall be reformed to the minimum extent necessary to make such provision valid and enforceable.
If Recipient institutes patent litigation against any entity (including a cross-claim or counterclaim in a lawsuit) alleging that the Program itself (excluding combinations of the Program with other software or hardware) infringes such Recipient's patent(s), then such Recipient's rights granted under Section 2(b) shall terminate as of the date such litigation is filed.
All Recipient's rights under this Agreement shall terminate if it fails to comply with any of the material terms or conditions of this Agreement and does not cure such failure in a reasonable period of time after becoming aware of such noncompliance. If all Recipient's rights under this Agreement terminate, Recipient agrees to cease use and distribution of the Program as soon as reasonably practicable. However, Recipient's obligations under this Agreement and any licenses granted by Recipient relating to the Program shall continue and survive.
Everyone is permitted to copy and distribute copies of this Agreement, but in order to avoid inconsistency the Agreement is copyrighted and may only be modified in the following manner. The Agreement Steward reserves the right to publish new versions (including revisions) of this Agreement from time to time. No one other than the Agreement Steward has the right to modify this Agreement. The Eclipse Foundation is the initial Agreement Steward. The Eclipse Foundation may assign the responsibility to serve as the Agreement Steward to a suitable separate entity. Each new version of the Agreement will be given a distinguishing version number. The Program (including Contributions) may always be distributed subject to the version of the Agreement under which it was received. In addition, after a new version of the Agreement is published, Contributor may elect to distribute the Program (including its Contributions) under the new version. Except as expressly stated in Sections 2(a) and 2(b) above, Recipient receives no rights or licenses to the intellectual property of any Contributor under this Agreement, whether expressly, by implication, estoppel or otherwise. All rights in the Program not expressly granted under this Agreement are reserved.
This Agreement is governed by the laws of the State of New York and the intellectual property laws of the United States of America. No party to this Agreement will bring a legal action under this Agreement more than one year after the cause of action arose. Each party waives its rights to a jury trial in any resulting litigation.

142
vendor/github.com/eclipse/paho.mqtt.golang/README.md generated vendored
View File

@ -1,32 +1,47 @@
[![GoDoc](https://godoc.org/github.com/eclipse/paho.mqtt.golang?status.svg)](https://godoc.org/github.com/eclipse/paho.mqtt.golang)
[![PkgGoDev](https://pkg.go.dev/badge/github.com/eclipse/paho.mqtt.golang)](https://pkg.go.dev/github.com/eclipse/paho.mqtt.golang)
[![Go Report Card](https://goreportcard.com/badge/github.com/eclipse/paho.mqtt.golang)](https://goreportcard.com/report/github.com/eclipse/paho.mqtt.golang)
Eclipse Paho MQTT Go client
===========================
This repository contains the source code for the [Eclipse Paho](http://eclipse.org/paho) MQTT Go client library.
This repository contains the source code for the [Eclipse Paho](https://eclipse.org/paho) MQTT 3.1/3.11 Go client library.
This code builds a library which enable applications to connect to an [MQTT](http://mqtt.org) broker to publish messages, and to subscribe to topics and receive published messages.
This code builds a library which enable applications to connect to an [MQTT](https://mqtt.org) broker to publish
messages, and to subscribe to topics and receive published messages.
This library supports a fully asynchronous mode of operation.
A client supporting MQTT V5 is [also available](https://github.com/eclipse/paho.golang).
Installation and Build
----------------------
This client is designed to work with the standard Go tools, so installation is as easy as:
The process depends upon whether you are using [modules](https://golang.org/ref/mod) (recommended) or `GOPATH`.
#### Modules
If you are using [modules](https://blog.golang.org/using-go-modules) then `import "github.com/eclipse/paho.mqtt.golang"`
and start using it. The necessary packages will be download automatically when you run `go build`.
Note that the latest release will be downloaded and changes may have been made since the release. If you have
encountered an issue, or wish to try the latest code for another reason, then run
`go get github.com/eclipse/paho.mqtt.golang@master` to get the latest commit.
#### GOPATH
Installation is as easy as:
```
go get github.com/eclipse/paho.mqtt.golang
```
The client depends on Google's [websockets](https://godoc.org/golang.org/x/net/websocket) and [proxy](https://godoc.org/golang.org/x/net/proxy) package,
also easily installed with the commands:
The client depends on Google's [proxy](https://godoc.org/golang.org/x/net/proxy) package and the
[websockets](https://godoc.org/github.com/gorilla/websocket) package, also easily installed with the commands:
```
go get golang.org/x/net/websocket
go get github.com/gorilla/websocket
go get golang.org/x/net/proxy
```
@ -35,27 +50,119 @@ Usage and API
-------------
Detailed API documentation is available by using to godoc tool, or can be browsed online
using the [godoc.org](http://godoc.org/github.com/eclipse/paho.mqtt.golang) service.
Make use of the library by importing it in your Go client source code. For example,
```
import "github.com/eclipse/paho.mqtt.golang"
```
using the [pkg.go.dev](https://pkg.go.dev/github.com/eclipse/paho.mqtt.golang) service.
Samples are available in the `cmd` directory for reference.
Note:
Runtime tracing
The library also supports using MQTT over websockets by using the `ws://` (unsecure) or `wss://` (secure) prefix in the
URI. If the client is running behind a corporate http/https proxy then the following environment variables `HTTP_PROXY`,
`HTTPS_PROXY` and `NO_PROXY` are taken into account when establishing the connection.
Troubleshooting
---------------
Tracing is enabled by assigning logs (from the Go log package) to the logging endpoints, ERROR, CRITICAL, WARN and DEBUG
If you are new to MQTT and your application is not working as expected reviewing the
[MQTT specification](https://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html), which this library implements,
is a good first step. [MQTT.org](https://mqtt.org) has some [good resources](https://mqtt.org/getting-started/) that answer many
common questions.
### Error Handling
The asynchronous nature of this library makes it easy to forget to check for errors. Consider using a go routine to
log these:
```go
t := client.Publish("topic", qos, retained, msg)
go func() {
_ = t.Wait() // Can also use '<-t.Done()' in releases > 1.2.0
if t.Error() != nil {
log.Error(t.Error()) // Use your preferred logging technique (or just fmt.Printf)
}
}()
```
### Logging
If you are encountering issues then enabling logging, both within this library and on your broker, is a good way to
begin troubleshooting. This library can produce various levels of log by assigning the logging endpoints, ERROR,
CRITICAL, WARN and DEBUG. For example:
```go
func main() {
mqtt.ERROR = log.New(os.Stdout, "[ERROR] ", 0)
mqtt.CRITICAL = log.New(os.Stdout, "[CRIT] ", 0)
mqtt.WARN = log.New(os.Stdout, "[WARN] ", 0)
mqtt.DEBUG = log.New(os.Stdout, "[DEBUG] ", 0)
// Connect, Subscribe, Publish etc..
}
```
### Common Problems
* Seemingly random disconnections may be caused by another client connecting to the broker with the same client
identifier; this is as per the [spec](https://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc384800405).
* Unless ordered delivery of messages is essential (and you have configured your broker to support this e.g.
`max_inflight_messages=1` in mosquitto) then set `ClientOptions.SetOrderMatters(false)`. Doing so will avoid the
below issue (deadlocks due to blocking message handlers).
* A `MessageHandler` (called when a new message is received) must not block (unless
`ClientOptions.SetOrderMatters(false)` set). If you wish to perform a long-running task, or publish a message, then
please use a go routine (blocking in the handler is a common cause of unexpected `pingresp
not received, disconnecting` errors).
* When QOS1+ subscriptions have been created previously and you connect with `CleanSession` set to false it is possible that the broker will deliver retained
messages before `Subscribe` can be called. To process these messages either configure a handler with `AddRoute` or
set a `DefaultPublishHandler`.
* Loss of network connectivity may not be detected immediately. If this is an issue then consider setting
`ClientOptions.KeepAlive` (sends regular messages to check the link is active).
* Brokers offer many configuration options; some settings may lead to unexpected results. If using Mosquitto check
`max_inflight_messages`, `max_queued_messages`, `persistence` (the defaults may not be what you expect).
Reporting bugs
--------------
Please report bugs by raising issues for this project in github https://github.com/eclipse/paho.mqtt.golang/issues
Please report bugs by raising issues for this project in github https://github.com/eclipse/paho.mqtt.golang/issues
*A limited number of contributors monitor the issues section so if you have a general question please consider the
resources in the [more information](#more-information) section (your question will be seen by more people, and you are
likely to receive an answer more quickly).*
We welcome bug reports, but it is important they are actionable. A significant percentage of issues reported are not
resolved due to a lack of information. If we cannot replicate the problem then it is unlikely we will be able to fix it.
The information required will vary from issue to issue but consider including:
* Which version of the package you are using (tag or commit - this should be in your go.mod file)
* A [Minimal, Reproducible Example](https://stackoverflow.com/help/minimal-reproducible-example). Providing an example
is the best way to demonstrate the issue you are facing; it is important this includes all relevant information
(including broker configuration). Docker (see `cmd/docker`) makes it relatively simple to provide a working end-to-end
example.
* A full, clear, description of the problem (detail what you are expecting vs what actually happens).
* Details of your attempts to resolve the issue (what have you tried, what worked, what did not).
* [Application Logs](#logging) covering the period the issue occurred. Unless you have isolated the root cause of the issue please include a link to a full log (including data from well before the problem arose).
* Broker Logs covering the period the issue occurred.
It is important to remember that this library does not stand alone; it communicates with a broker and any issues you are
seeing may be due to:
* Bugs in your code.
* Bugs in this library.
* The broker configuration.
* Bugs in the broker.
* Issues with whatever you are communicating with.
When submitting an issue, please ensure that you provide sufficient details to enable us to eliminate causes outside of
this library.
Contributing
------------
We welcome pull requests but before your contribution can be accepted by the project, you need to create and
electronically sign the Eclipse Contributor Agreement (ECA) and sign off on the Eclipse Foundation Certificate of Origin.
More information is available in the
[Eclipse Development Resources](http://wiki.eclipse.org/Development_Resources/Contributing_via_Git); please take special
note of the requirement that the commit record contain a "Signed-off-by" entry.
More information
----------------
@ -65,3 +172,6 @@ Discussion of the Paho clients takes place on the [Eclipse paho-dev mailing list
General questions about the MQTT protocol are discussed in the [MQTT Google Group](https://groups.google.com/forum/?hl=en-US&fromgroups#!forum/mqtt).
There is much more information available via the [MQTT community site](http://mqtt.org).
[Stack Overflow](https://stackoverflow.com/questions/tagged/mqtt+go) has a range questions covering a range of common
issues (both relating to use of this library and MQTT in general).

1022
vendor/github.com/eclipse/paho.mqtt.golang/client.go generated vendored
View File

File diff suppressed because it is too large Load Diff

1
vendor/github.com/eclipse/paho.mqtt.golang/components.go generated vendored
View File

@ -28,4 +28,5 @@ const (
TST component = "[test] "
STA component = "[state] "
ERR component = "[error] "
ROU component = "[router] "
)

14
vendor/github.com/eclipse/paho.mqtt.golang/filestore.go generated vendored
View File

@ -101,7 +101,7 @@ func (store *FileStore) Get(key string) packets.ControlPacket {
store.RLock()
defer store.RUnlock()
if !store.opened {
ERROR.Println(STR, "Trying to use file store, but not open")
ERROR.Println(STR, "trying to use file store, but not open")
return nil
}
filepath := fullpath(store.directory, key)
@ -117,14 +117,16 @@ func (store *FileStore) Get(key string) packets.ControlPacket {
if rerr != nil {
newpath := corruptpath(store.directory, key)
WARN.Println(STR, "corrupted file detected:", rerr.Error(), "archived at:", newpath)
os.Rename(filepath, newpath)
if err := os.Rename(filepath, newpath); err != nil {
ERROR.Println(STR, err)
}
return nil
}
return msg
}
// All will provide a list of all of the keys associated with messages
// currenly residing in the FileStore.
// currently residing in the FileStore.
func (store *FileStore) All() []string {
store.RLock()
defer store.RUnlock()
@ -156,7 +158,7 @@ func (store *FileStore) all() []string {
var files fileInfos
if !store.opened {
ERROR.Println(STR, "Trying to use file store, but not open")
ERROR.Println(STR, "trying to use file store, but not open")
return nil
}
@ -166,7 +168,7 @@ func (store *FileStore) all() []string {
for _, f := range files {
DEBUG.Println(STR, "file in All():", f.Name())
name := f.Name()
if name[len(name)-4:len(name)] != msgExt {
if name[len(name)-4:] != msgExt {
DEBUG.Println(STR, "skipping file, doesn't have right extension: ", name)
continue
}
@ -179,7 +181,7 @@ func (store *FileStore) all() []string {
// lockless
func (store *FileStore) del(key string) {
if !store.opened {
ERROR.Println(STR, "Trying to use file store, but not open")
ERROR.Println(STR, "trying to use file store, but not open")
return
}
DEBUG.Println(STR, "store del filepath:", store.directory)

2
vendor/github.com/eclipse/paho.mqtt.golang/memstore.go generated vendored
View File

@ -88,7 +88,7 @@ func (store *MemoryStore) All() []string {
ERROR.Println(STR, "Trying to use memory store, but not open")
return nil
}
keys := []string{}
var keys []string
for k := range store.messages {
keys = append(keys, k)
}

4
vendor/github.com/eclipse/paho.mqtt.golang/message.go generated vendored
View File

@ -16,9 +16,9 @@ package mqtt
import (
"net/url"
"sync"
"github.com/eclipse/paho.mqtt.golang/packets"
"sync"
)
// Message defines the externals that a message implementation must support
@ -114,7 +114,7 @@ func newConnectMsgFromOptions(options *ClientOptions, broker *url.URL) *packets.
if username != "" {
m.UsernameFlag = true
m.Username = username
//mustn't have password without user as well
// mustn't have password without user as well
if password != "" {
m.PasswordFlag = true
m.Password = []byte(password)

69
vendor/github.com/eclipse/paho.mqtt.golang/messageids.go generated vendored
View File

@ -28,6 +28,8 @@ type MId uint16
type messageIds struct {
sync.RWMutex
index map[uint16]tokenCompletor
lastIssuedID uint16 // The most recently issued ID. Used so we cycle through ids rather than immediately reusing them (can make debugging easier)
}
const (
@ -40,11 +42,11 @@ func (mids *messageIds) cleanUp() {
for _, token := range mids.index {
switch token.(type) {
case *PublishToken:
token.setError(fmt.Errorf("Connection lost before Publish completed"))
token.setError(fmt.Errorf("connection lost before Publish completed"))
case *SubscribeToken:
token.setError(fmt.Errorf("Connection lost before Subscribe completed"))
token.setError(fmt.Errorf("connection lost before Subscribe completed"))
case *UnsubscribeToken:
token.setError(fmt.Errorf("Connection lost before Unsubscribe completed"))
token.setError(fmt.Errorf("connection lost before Unsubscribe completed"))
case nil:
continue
}
@ -71,18 +73,33 @@ func (mids *messageIds) claimID(token tokenCompletor, id uint16) {
old.flowComplete()
mids.index[id] = token
}
if id > mids.lastIssuedID {
mids.lastIssuedID = id
}
}
// getID will return an available id or 0 if none available
// The id will generally be the previous id + 1 (because this makes tracing messages a bit simpler)
func (mids *messageIds) getID(t tokenCompletor) uint16 {
mids.Lock()
defer mids.Unlock()
for i := midMin; i < midMax; i++ {
i := mids.lastIssuedID // note: the only situation where lastIssuedID is 0 the map will be empty
looped := false // uint16 will loop from 65535->0
for {
i++
if i == 0 { // skip 0 because its not a valid id (Control Packets MUST contain a non-zero 16-bit Packet Identifier [MQTT-2.3.1-1])
i++
looped = true
}
if _, ok := mids.index[i]; !ok {
mids.index[i] = t
mids.lastIssuedID = i
return i
}
if (looped && i == mids.lastIssuedID) || (mids.lastIssuedID == 0 && i == midMax) { // lastIssuedID will be 0 at startup
return 0 // no free ids
}
}
return 0
}
func (mids *messageIds) getToken(id uint16) tokenCompletor {
@ -98,14 +115,23 @@ type DummyToken struct {
id uint16
}
// Wait implements the Token Wait method.
func (d *DummyToken) Wait() bool {
return true
}
// WaitTimeout implements the Token WaitTimeout method.
func (d *DummyToken) WaitTimeout(t time.Duration) bool {
return true
}
// Done implements the Token Done method.
func (d *DummyToken) Done() <-chan struct{} {
ch := make(chan struct{})
close(ch)
return ch
}
func (d *DummyToken) flowComplete() {
ERROR.Printf("A lookup for token %d returned nil\n", d.id)
}
@ -115,3 +141,36 @@ func (d *DummyToken) Error() error {
}
func (d *DummyToken) setError(e error) {}
// PlaceHolderToken does nothing and was implemented to allow a messageid to be reserved
// it differs from DummyToken in that calling flowComplete does not generate an error (it
// is expected that flowComplete will be called when the token is overwritten with a real token)
type PlaceHolderToken struct {
id uint16
}
// Wait implements the Token Wait method.
func (p *PlaceHolderToken) Wait() bool {
return true
}
// WaitTimeout implements the Token WaitTimeout method.
func (p *PlaceHolderToken) WaitTimeout(t time.Duration) bool {
return true
}
// Done implements the Token Done method.
func (p *PlaceHolderToken) Done() <-chan struct{} {
ch := make(chan struct{})
close(ch)
return ch
}
func (p *PlaceHolderToken) flowComplete() {
}
func (p *PlaceHolderToken) Error() error {
return nil
}
func (p *PlaceHolderToken) setError(e error) {}

633
vendor/github.com/eclipse/paho.mqtt.golang/net.go generated vendored
View File

@ -15,341 +15,450 @@
package mqtt
import (
"crypto/tls"
"errors"
"fmt"
"io"
"net"
"net/http"
"net/url"
"os"
"reflect"
"sync/atomic"
"strings"
"sync"
"time"
"github.com/eclipse/paho.mqtt.golang/packets"
"golang.org/x/net/proxy"
"golang.org/x/net/websocket"
)
func signalError(c chan<- error, err error) {
select {
case c <- err:
const closedNetConnErrorText = "use of closed network connection" // error string for closed conn (https://golang.org/src/net/error_test.go)
// ConnectMQTT takes a connected net.Conn and performs the initial MQTT handshake. Parameters are:
// conn - Connected net.Conn
// cm - Connect Packet with everything other than the protocol name/version populated (historical reasons)
// protocolVersion - The protocol version to attempt to connect with
//
// Note that, for backward compatibility, ConnectMQTT() suppresses the actual connection error (compare to connectMQTT()).
func ConnectMQTT(conn net.Conn, cm *packets.ConnectPacket, protocolVersion uint) (byte, bool) {
rc, sessionPresent, _ := connectMQTT(conn, cm, protocolVersion)
return rc, sessionPresent
}
func connectMQTT(conn io.ReadWriter, cm *packets.ConnectPacket, protocolVersion uint) (byte, bool, error) {
switch protocolVersion {
case 3:
DEBUG.Println(CLI, "Using MQTT 3.1 protocol")
cm.ProtocolName = "MQIsdp"
cm.ProtocolVersion = 3
case 0x83:
DEBUG.Println(CLI, "Using MQTT 3.1b protocol")
cm.ProtocolName = "MQIsdp"
cm.ProtocolVersion = 0x83
case 0x84:
DEBUG.Println(CLI, "Using MQTT 3.1.1b protocol")
cm.ProtocolName = "MQTT"
cm.ProtocolVersion = 0x84
default:
DEBUG.Println(CLI, "Using MQTT 3.1.1 protocol")
cm.ProtocolName = "MQTT"
cm.ProtocolVersion = 4
}
if err := cm.Write(conn); err != nil {
ERROR.Println(CLI, err)
return packets.ErrNetworkError, false, err
}
rc, sessionPresent, err := verifyCONNACK(conn)
return rc, sessionPresent, err
}
func openConnection(uri *url.URL, tlsc *tls.Config, timeout time.Duration, headers http.Header) (net.Conn, error) {
switch uri.Scheme {
case "ws":
config, _ := websocket.NewConfig(uri.String(), fmt.Sprintf("http://%s", uri.Host))
config.Protocol = []string{"mqtt"}
config.Header = headers
config.Dialer = &net.Dialer{Timeout: timeout}
conn, err := websocket.DialConfig(config)
if err != nil {
return nil, err
}
conn.PayloadType = websocket.BinaryFrame
return conn, err
case "wss":
config, _ := websocket.NewConfig(uri.String(), fmt.Sprintf("https://%s", uri.Host))
config.Protocol = []string{"mqtt"}
config.TlsConfig = tlsc
config.Header = headers
config.Dialer = &net.Dialer{Timeout: timeout}
conn, err := websocket.DialConfig(config)
if err != nil {
return nil, err
}
conn.PayloadType = websocket.BinaryFrame
return conn, err
case "tcp":
allProxy := os.Getenv("all_proxy")
if len(allProxy) == 0 {
conn, err := net.DialTimeout("tcp", uri.Host, timeout)
if err != nil {
return nil, err
}
return conn, nil
}
proxyDialer := proxy.FromEnvironment()
// This function is only used for receiving a connack
// when the connection is first started.
// This prevents receiving incoming data while resume
// is in progress if clean session is false.
func verifyCONNACK(conn io.Reader) (byte, bool, error) {
DEBUG.Println(NET, "connect started")
conn, err := proxyDialer.Dial("tcp", uri.Host)
if err != nil {
return nil, err
}
return conn, nil
case "unix":
conn, err := net.DialTimeout("unix", uri.Host, timeout)
if err != nil {
return nil, err
}
return conn, nil
case "ssl":
fallthrough
case "tls":
fallthrough
case "tcps":
allProxy := os.Getenv("all_proxy")
if len(allProxy) == 0 {
conn, err := tls.DialWithDialer(&net.Dialer{Timeout: timeout}, "tcp", uri.Host, tlsc)
if err != nil {
return nil, err
}
return conn, nil
}
proxyDialer := proxy.FromEnvironment()
conn, err := proxyDialer.Dial("tcp", uri.Host)
if err != nil {
return nil, err
}
tlsConn := tls.Client(conn, tlsc)
err = tlsConn.Handshake()
if err != nil {
conn.Close()
return nil, err
}
return tlsConn, nil
ca, err := packets.ReadPacket(conn)
if err != nil {
ERROR.Println(NET, "connect got error", err)
return packets.ErrNetworkError, false, err
}
return nil, errors.New("Unknown protocol")
if ca == nil {
ERROR.Println(NET, "received nil packet")
return packets.ErrNetworkError, false, errors.New("nil CONNACK packet")
}
msg, ok := ca.(*packets.ConnackPacket)
if !ok {
ERROR.Println(NET, "received msg that was not CONNACK")
return packets.ErrNetworkError, false, errors.New("non-CONNACK first packet received")
}
DEBUG.Println(NET, "received connack")
return msg.ReturnCode, msg.SessionPresent, nil
}
// actually read incoming messages off the wire
// send Message object into ibound channel
func incoming(c *client) {
// inbound encapsulates the output from startIncoming.
// err - If != nil then an error has occurred
// cp - A control packet received over the network link
type inbound struct {
err error
cp packets.ControlPacket
}
// startIncoming initiates a goroutine that reads incoming messages off the wire and sends them to the channel (returned).
// If there are any issues with the network connection then the returned channel will be closed and the goroutine will exit
// (so closing the connection will terminate the goroutine)
func startIncoming(conn io.Reader) <-chan inbound {
var err error
var cp packets.ControlPacket
defer c.workers.Done()
ibound := make(chan inbound)
DEBUG.Println(NET, "incoming started")
for {
if cp, err = packets.ReadPacket(c.conn); err != nil {
break
}
DEBUG.Println(NET, "Received Message")
select {
case c.ibound <- cp:
// Notify keepalive logic that we recently received a packet
if c.options.KeepAlive != 0 {
c.lastReceived.Store(time.Now())
}
case <-c.stop:
// This avoids a deadlock should a message arrive while shutting down.
// In that case the "reader" of c.ibound might already be gone
WARN.Println(NET, "incoming dropped a received message during shutdown")
break
}
}
// We received an error on read.
// If disconnect is in progress, swallow error and return
select {
case <-c.stop:
DEBUG.Println(NET, "incoming stopped")
return
// Not trying to disconnect, send the error to the errors channel
default:
ERROR.Println(NET, "incoming stopped with error", err)
signalError(c.errors, err)
return
}
}
// receive a Message object on obound, and then
// actually send outgoing message to the wire
func outgoing(c *client) {
defer c.workers.Done()
DEBUG.Println(NET, "outgoing started")
for {
DEBUG.Println(NET, "outgoing waiting for an outbound message")
select {
case <-c.stop:
DEBUG.Println(NET, "outgoing stopped")
return
case pub := <-c.obound:
msg := pub.p.(*packets.PublishPacket)
if c.options.WriteTimeout > 0 {
c.conn.SetWriteDeadline(time.Now().Add(c.options.WriteTimeout))
}
if err := msg.Write(c.conn); err != nil {
ERROR.Println(NET, "outgoing stopped with error", err)
pub.t.setError(err)
signalError(c.errors, err)
return
}
if c.options.WriteTimeout > 0 {
// If we successfully wrote, we don't want the timeout to happen during an idle period
// so we reset it to infinite.
c.conn.SetWriteDeadline(time.Time{})
}
if msg.Qos == 0 {
pub.t.flowComplete()
}
DEBUG.Println(NET, "obound wrote msg, id:", msg.MessageID)
case msg := <-c.oboundP:
switch msg.p.(type) {
case *packets.SubscribePacket:
msg.p.(*packets.SubscribePacket).MessageID = c.getID(msg.t)
case *packets.UnsubscribePacket:
msg.p.(*packets.UnsubscribePacket).MessageID = c.getID(msg.t)
}
DEBUG.Println(NET, "obound priority msg to write, type", reflect.TypeOf(msg.p))
if err := msg.p.Write(c.conn); err != nil {
ERROR.Println(NET, "outgoing stopped with error", err)
if msg.t != nil {
msg.t.setError(err)
go func() {
for {
if cp, err = packets.ReadPacket(conn); err != nil {
// We do not want to log the error if it is due to the network connection having been closed
// elsewhere (i.e. after sending DisconnectPacket). Detecting this situation is the subject of
// https://github.com/golang/go/issues/4373
if !strings.Contains(err.Error(), closedNetConnErrorText) {
ibound <- inbound{err: err}
}
signalError(c.errors, err)
return
}
switch msg.p.(type) {
case *packets.DisconnectPacket:
msg.t.(*DisconnectToken).flowComplete()
DEBUG.Println(NET, "outbound wrote disconnect, stopping")
close(ibound)
DEBUG.Println(NET, "incoming complete")
return
}
DEBUG.Println(NET, "startIncoming Received Message")
ibound <- inbound{cp: cp}
}
// Reset ping timer after sending control packet.
if c.options.KeepAlive != 0 {
c.lastSent.Store(time.Now())
}
}
}()
return ibound
}
// receive Message objects on ibound
// store messages if necessary
// send replies on obound
// delete messages from store if necessary
func alllogic(c *client) {
defer c.workers.Done()
DEBUG.Println(NET, "logic started")
// incomingComms encapsulates the possible output of the incomingComms routine. If err != nil then an error has occurred and
// the routine will have terminated; otherwise one of the other members should be non-nil
type incomingComms struct {
err error // If non-nil then there has been an error (ignore everything else)
outbound *PacketAndToken // Packet (with token) than needs to be sent out (e.g. an acknowledgement)
incomingPub *packets.PublishPacket // A new publish has been received; this will need to be passed on to our user
}
for {
DEBUG.Println(NET, "logic waiting for msg on ibound")
// startIncomingComms initiates incoming communications; this includes starting a goroutine to process incoming
// messages.
// Accepts a channel of inbound messages from the store (persisted messages); note this must be closed as soon as the
// everything in the store has been sent.
// Returns a channel that will be passed any received packets; this will be closed on a network error (and inboundFromStore closed)
func startIncomingComms(conn io.Reader,
c commsFns,
inboundFromStore <-chan packets.ControlPacket,
) <-chan incomingComms {
ibound := startIncoming(conn) // Start goroutine that reads from network connection
output := make(chan incomingComms)
DEBUG.Println(NET, "startIncomingComms started")
go func() {
for {
if inboundFromStore == nil && ibound == nil {
close(output)
DEBUG.Println(NET, "startIncomingComms goroutine complete")
return // As soon as ibound is closed we can exit (should have already processed an error)
}
DEBUG.Println(NET, "logic waiting for msg on ibound")
var msg packets.ControlPacket
var ok bool
select {
case msg, ok = <-inboundFromStore:
if !ok {
DEBUG.Println(NET, "startIncomingComms: inboundFromStore complete")
inboundFromStore = nil // should happen quickly as this is only for persisted messages
continue
}
DEBUG.Println(NET, "startIncomingComms: got msg from store")
case ibMsg, ok := <-ibound:
if !ok {
DEBUG.Println(NET, "startIncomingComms: ibound complete")
ibound = nil
continue
}
DEBUG.Println(NET, "startIncomingComms: got msg on ibound")
// If the inbound comms routine encounters any issues it will send us an error.
if ibMsg.err != nil {
output <- incomingComms{err: ibMsg.err}
continue // Usually the channel will be closed immediately after sending an error but safer that we do not assume this
}
msg = ibMsg.cp
c.persistInbound(msg)
c.UpdateLastReceived() // Notify keepalive logic that we recently received a packet
}
select {
case msg := <-c.ibound:
DEBUG.Println(NET, "logic got msg on ibound")
persistInbound(c.persist, msg)
switch m := msg.(type) {
case *packets.PingrespPacket:
DEBUG.Println(NET, "received pingresp")
atomic.StoreInt32(&c.pingOutstanding, 0)
DEBUG.Println(NET, "startIncomingComms: received pingresp")
c.pingRespReceived()
case *packets.SubackPacket:
DEBUG.Println(NET, "received suback, id:", m.MessageID)
DEBUG.Println(NET, "startIncomingComms: received suback, id:", m.MessageID)
token := c.getToken(m.MessageID)
switch t := token.(type) {
case *SubscribeToken:
DEBUG.Println(NET, "granted qoss", m.ReturnCodes)
if t, ok := token.(*SubscribeToken); ok {
DEBUG.Println(NET, "startIncomingComms: granted qoss", m.ReturnCodes)
for i, qos := range m.ReturnCodes {
t.subResult[t.subs[i]] = qos
}
}
token.flowComplete()
c.freeID(m.MessageID)
case *packets.UnsubackPacket:
DEBUG.Println(NET, "received unsuback, id:", m.MessageID)
DEBUG.Println(NET, "startIncomingComms: received unsuback, id:", m.MessageID)
c.getToken(m.MessageID).flowComplete()
c.freeID(m.MessageID)
case *packets.PublishPacket:
DEBUG.Println(NET, "received publish, msgId:", m.MessageID)
DEBUG.Println(NET, "putting msg on onPubChan")
switch m.Qos {
case 2:
c.incomingPubChan <- m
DEBUG.Println(NET, "done putting msg on incomingPubChan")
case 1:
c.incomingPubChan <- m
DEBUG.Println(NET, "done putting msg on incomingPubChan")
case 0:
select {
case c.incomingPubChan <- m:
case <-c.stop:
}
DEBUG.Println(NET, "done putting msg on incomingPubChan")
}
DEBUG.Println(NET, "startIncomingComms: received publish, msgId:", m.MessageID)
output <- incomingComms{incomingPub: m}
case *packets.PubackPacket:
DEBUG.Println(NET, "received puback, id:", m.MessageID)
// c.receipts.get(msg.MsgId()) <- Receipt{}
// c.receipts.end(msg.MsgId())
DEBUG.Println(NET, "startIncomingComms: received puback, id:", m.MessageID)
c.getToken(m.MessageID).flowComplete()
c.freeID(m.MessageID)
case *packets.PubrecPacket:
DEBUG.Println(NET, "received pubrec, id:", m.MessageID)
DEBUG.Println(NET, "startIncomingComms: received pubrec, id:", m.MessageID)
prel := packets.NewControlPacket(packets.Pubrel).(*packets.PubrelPacket)
prel.MessageID = m.MessageID
select {
case c.oboundP <- &PacketAndToken{p: prel, t: nil}:
case <-c.stop:
}
output <- incomingComms{outbound: &PacketAndToken{p: prel, t: nil}}
case *packets.PubrelPacket:
DEBUG.Println(NET, "received pubrel, id:", m.MessageID)
DEBUG.Println(NET, "startIncomingComms: received pubrel, id:", m.MessageID)
pc := packets.NewControlPacket(packets.Pubcomp).(*packets.PubcompPacket)
pc.MessageID = m.MessageID
persistOutbound(c.persist, pc)
select {
case c.oboundP <- &PacketAndToken{p: pc, t: nil}:
case <-c.stop:
}
c.persistOutbound(pc)
output <- incomingComms{outbound: &PacketAndToken{p: pc, t: nil}}
case *packets.PubcompPacket:
DEBUG.Println(NET, "received pubcomp, id:", m.MessageID)
DEBUG.Println(NET, "startIncomingComms: received pubcomp, id:", m.MessageID)
c.getToken(m.MessageID).flowComplete()
c.freeID(m.MessageID)
}
case <-c.stop:
WARN.Println(NET, "logic stopped")
return
}
}
}()
return output
}
func (c *client) ackFunc(packet *packets.PublishPacket) func() {
// startOutgoingComms initiates a go routine to transmit outgoing packets.
// Pass in an open network connection and channels for outbound messages (including those triggered
// directly from incoming comms).
// Returns a channel that will receive details of any errors (closed when the goroutine exits)
// This function wil only terminate when all input channels are closed
func startOutgoingComms(conn net.Conn,
c commsFns,
oboundp <-chan *PacketAndToken,
obound <-chan *PacketAndToken,
oboundFromIncoming <-chan *PacketAndToken,
) <-chan error {
errChan := make(chan error)
DEBUG.Println(NET, "outgoing started")
go func() {
for {
DEBUG.Println(NET, "outgoing waiting for an outbound message")
// This goroutine will only exits when all of the input channels we receive on have been closed. This approach is taken to avoid any
// deadlocks (if the connection goes down there are limited options as to what we can do with anything waiting on us and
// throwing away the packets seems the best option)
if oboundp == nil && obound == nil && oboundFromIncoming == nil {
DEBUG.Println(NET, "outgoing comms stopping")
close(errChan)
return
}
select {
case pub, ok := <-obound:
if !ok {
obound = nil
continue
}
msg := pub.p.(*packets.PublishPacket)
DEBUG.Println(NET, "obound msg to write", msg.MessageID)
writeTimeout := c.getWriteTimeOut()
if writeTimeout > 0 {
if err := conn.SetWriteDeadline(time.Now().Add(writeTimeout)); err != nil {
ERROR.Println(NET, "SetWriteDeadline ", err)
}
}
if err := msg.Write(conn); err != nil {
ERROR.Println(NET, "outgoing obound reporting error ", err)
pub.t.setError(err)
// report error if it's not due to the connection being closed elsewhere
if !strings.Contains(err.Error(), closedNetConnErrorText) {
errChan <- err
}
continue
}
if writeTimeout > 0 {
// If we successfully wrote, we don't want the timeout to happen during an idle period
// so we reset it to infinite.
if err := conn.SetWriteDeadline(time.Time{}); err != nil {
ERROR.Println(NET, "SetWriteDeadline to 0 ", err)
}
}
if msg.Qos == 0 {
pub.t.flowComplete()
}
DEBUG.Println(NET, "obound wrote msg, id:", msg.MessageID)
case msg, ok := <-oboundp:
if !ok {
oboundp = nil
continue
}
DEBUG.Println(NET, "obound priority msg to write, type", reflect.TypeOf(msg.p))
if err := msg.p.Write(conn); err != nil {
ERROR.Println(NET, "outgoing oboundp reporting error ", err)
if msg.t != nil {
msg.t.setError(err)
}
errChan <- err
continue
}
if _, ok := msg.p.(*packets.DisconnectPacket); ok {
msg.t.(*DisconnectToken).flowComplete()
DEBUG.Println(NET, "outbound wrote disconnect, closing connection")
// As per the MQTT spec "After sending a DISCONNECT Packet the Client MUST close the Network Connection"
// Closing the connection will cause the goroutines to end in sequence (starting with incoming comms)
conn.Close()
}
case msg, ok := <-oboundFromIncoming: // message triggered by an inbound message (PubrecPacket or PubrelPacket)
if !ok {
oboundFromIncoming = nil
continue
}
DEBUG.Println(NET, "obound from incoming msg to write, type", reflect.TypeOf(msg.p), " ID ", msg.p.Details().MessageID)
if err := msg.p.Write(conn); err != nil {
ERROR.Println(NET, "outgoing oboundFromIncoming reporting error", err)
if msg.t != nil {
msg.t.setError(err)
}
errChan <- err
continue
}
}
c.UpdateLastSent() // Record that a packet has been received (for keepalive routine)
}
}()
return errChan
}
// commsFns provide access to the client state (messageids, requesting disconnection and updating timing)
type commsFns interface {
getToken(id uint16) tokenCompletor // Retrieve the token for the specified messageid (if none then a dummy token must be returned)
freeID(id uint16) // Release the specified messageid (clearing out of any persistent store)
UpdateLastReceived() // Must be called whenever a packet is received
UpdateLastSent() // Must be called whenever a packet is successfully sent
getWriteTimeOut() time.Duration // Return the writetimeout (or 0 if none)
persistOutbound(m packets.ControlPacket) // add the packet to the outbound store
persistInbound(m packets.ControlPacket) // add the packet to the inbound store
pingRespReceived() // Called when a ping response is received
}
// startComms initiates goroutines that handles communications over the network connection
// Messages will be stored (via commsFns) and deleted from the store as necessary
// It returns two channels:
// packets.PublishPacket - Will receive publish packets received over the network.
// Closed when incoming comms routines exit (on shutdown or if network link closed)
// error - Any errors will be sent on this channel. The channel is closed when all comms routines have shut down
//
// Note: The comms routines monitoring oboundp and obound will not shutdown until those channels are both closed. Any messages received between the
// connection being closed and those channels being closed will generate errors (and nothing will be sent). That way the chance of a deadlock is
// minimised.
func startComms(conn net.Conn, // Network connection (must be active)
c commsFns, // getters and setters to enable us to cleanly interact with client
inboundFromStore <-chan packets.ControlPacket, // Inbound packets from the persistence store (should be closed relatively soon after startup)
oboundp <-chan *PacketAndToken,
obound <-chan *PacketAndToken) (
<-chan *packets.PublishPacket, // Publishpackages received over the network
<-chan error, // Any errors (should generally trigger a disconnect)
) {
// Start inbound comms handler; this needs to be able to transmit messages so we start a go routine to add these to the priority outbound channel
ibound := startIncomingComms(conn, c, inboundFromStore)
outboundFromIncoming := make(chan *PacketAndToken) // Will accept outgoing messages triggered by startIncomingComms (e.g. acknowledgements)
// Start the outgoing handler. It is important to note that output from startIncomingComms is fed into startOutgoingComms (for ACK's)
oboundErr := startOutgoingComms(conn, c, oboundp, obound, outboundFromIncoming)
DEBUG.Println(NET, "startComms started")
// Run up go routines to handle the output from the above comms functions - these are handled in separate
// go routines because they can interact (e.g. ibound triggers an ACK to obound which triggers an error)
var wg sync.WaitGroup
wg.Add(2)
outPublish := make(chan *packets.PublishPacket)
outError := make(chan error)
// Any messages received get passed to the appropriate channel
go func() {
for ic := range ibound {
if ic.err != nil {
outError <- ic.err
continue
}
if ic.outbound != nil {
outboundFromIncoming <- ic.outbound
continue
}
if ic.incomingPub != nil {
outPublish <- ic.incomingPub
continue
}
ERROR.Println(STR, "startComms received empty incomingComms msg")
}
// Close channels that will not be written to again (allowing other routines to exit)
close(outboundFromIncoming)
close(outPublish)
wg.Done()
}()
// Any errors will be passed out to our caller
go func() {
for err := range oboundErr {
outError <- err
}
wg.Done()
}()
// outError is used by both routines so can only be closed when they are both complete
go func() {
wg.Wait()
close(outError)
DEBUG.Println(NET, "startComms closing outError")
}()
return outPublish, outError
}
// ackFunc acknowledges a packet
// WARNING the function returned must not be called if the comms routine is shutting down or not running
// (it needs outgoing comms in order to send the acknowledgement). Currently this is only called from
// matchAndDispatch which will be shutdown before the comms are
func ackFunc(oboundP chan *PacketAndToken, persist Store, packet *packets.PublishPacket) func() {
return func() {
switch packet.Qos {
case 2:
pr := packets.NewControlPacket(packets.Pubrec).(*packets.PubrecPacket)
pr.MessageID = packet.MessageID
DEBUG.Println(NET, "putting pubrec msg on obound")
select {
case c.oboundP <- &PacketAndToken{p: pr, t: nil}:
case <-c.stop:
}
oboundP <- &PacketAndToken{p: pr, t: nil}
DEBUG.Println(NET, "done putting pubrec msg on obound")
case 1:
pa := packets.NewControlPacket(packets.Puback).(*packets.PubackPacket)
pa.MessageID = packet.MessageID
DEBUG.Println(NET, "putting puback msg on obound")
persistOutbound(c.persist, pa)
select {
case c.oboundP <- &PacketAndToken{p: pa, t: nil}:
case <-c.stop:
}
persistOutbound(persist, pa)
oboundP <- &PacketAndToken{p: pa, t: nil}
DEBUG.Println(NET, "done putting puback msg on obound")
case 0:
// do nothing, since there is no need to send an ack packet back
}
}
}
func errorWatch(c *client) {
defer c.workers.Done()
select {
case <-c.stop:
WARN.Println(NET, "errorWatch stopped")
return
case err := <-c.errors:
ERROR.Println(NET, "error triggered, stopping")
go c.internalConnLost(err)
return
}
}

92
vendor/github.com/eclipse/paho.mqtt.golang/netconn.go generated vendored Normal file
View File

@ -0,0 +1,92 @@
/*
* Copyright (c) 2013 IBM Corp.
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* which accompanies this distribution, and is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* Contributors:
* Seth Hoenig
* Allan Stockdill-Mander
* Mike Robertson
*/
package mqtt
import (
"crypto/tls"
"errors"
"net"
"net/http"
"net/url"
"os"
"time"
"golang.org/x/net/proxy"
)
//
// This just establishes the network connection; once established the type of connection should be irrelevant
//
// openConnection opens a network connection using the protocol indicated in the URL.
// Does not carry out any MQTT specific handshakes.
func openConnection(uri *url.URL, tlsc *tls.Config, timeout time.Duration, headers http.Header, websocketOptions *WebsocketOptions) (net.Conn, error) {
switch uri.Scheme {
case "ws":
conn, err := NewWebsocket(uri.String(), nil, timeout, headers, websocketOptions)
return conn, err
case "wss":
conn, err := NewWebsocket(uri.String(), tlsc, timeout, headers, websocketOptions)
return conn, err
case "mqtt", "tcp":
allProxy := os.Getenv("all_proxy")
if len(allProxy) == 0 {
conn, err := net.DialTimeout("tcp", uri.Host, timeout)
if err != nil {
return nil, err
}
return conn, nil
}
proxyDialer := proxy.FromEnvironment()
conn, err := proxyDialer.Dial("tcp", uri.Host)
if err != nil {
return nil, err
}
return conn, nil
case "unix":
conn, err := net.DialTimeout("unix", uri.Host, timeout)
if err != nil {
return nil, err
}
return conn, nil
case "ssl", "tls", "mqtts", "mqtt+ssl", "tcps":
allProxy := os.Getenv("all_proxy")
if len(allProxy) == 0 {
conn, err := tls.DialWithDialer(&net.Dialer{Timeout: timeout}, "tcp", uri.Host, tlsc)
if err != nil {
return nil, err
}
return conn, nil
}
proxyDialer := proxy.FromEnvironment()
conn, err := proxyDialer.Dial("tcp", uri.Host)
if err != nil {
return nil, err
}
tlsConn := tls.Client(conn, tlsc)
err = tlsConn.Handshake()
if err != nil {
_ = conn.Close()
return nil, err
}
return tlsConn, nil
}
return nil, errors.New("unknown protocol")
}

93
vendor/github.com/eclipse/paho.mqtt.golang/options.go generated vendored
View File

@ -10,6 +10,7 @@
* Seth Hoenig
* Allan Stockdill-Mander
* Mike Robertson
* Måns Ansgariusson
*/
// Portions copyright © 2018 TIBCO Software Inc.
@ -44,7 +45,15 @@ type ConnectionLostHandler func(Client, error)
// at initial connection and on reconnection
type OnConnectHandler func(Client)
// ClientOptions contains configurable options for an Client.
// ReconnectHandler is invoked prior to reconnecting after
// the initial connection is lost
type ReconnectHandler func(Client, *ClientOptions)
// ConnectionAttemptHandler is invoked prior to making the initial connection.
type ConnectionAttemptHandler func(broker *url.URL, tlsCfg *tls.Config) *tls.Config
// ClientOptions contains configurable options for an Client. Note that these should be set using the
// relevant methods (e.g. AddBroker) rather than directly. See those functions for information on usage.
type ClientOptions struct {
Servers []*url.URL
ClientID string
@ -66,21 +75,26 @@ type ClientOptions struct {
ConnectTimeout time.Duration
MaxReconnectInterval time.Duration
AutoReconnect bool
ConnectRetryInterval time.Duration
ConnectRetry bool
Store Store
DefaultPublishHandler MessageHandler
OnConnect OnConnectHandler
OnConnectionLost ConnectionLostHandler
OnReconnecting ReconnectHandler
OnConnectAttempt ConnectionAttemptHandler
WriteTimeout time.Duration
MessageChannelDepth uint
ResumeSubs bool
HTTPHeaders http.Header
WebsocketOptions *WebsocketOptions
}
// NewClientOptions will create a new ClientClientOptions type with some
// default values.
// Port: 1883
// CleanSession: True
// Order: True
// Order: True (note: it is recommended that this be set to FALSE unless order is important)
// KeepAlive: 30 (seconds)
// ConnectTimeout: 30 (seconds)
// MaxReconnectInterval 10 (minutes)
@ -105,13 +119,16 @@ func NewClientOptions() *ClientOptions {
ConnectTimeout: 30 * time.Second,
MaxReconnectInterval: 10 * time.Minute,
AutoReconnect: true,
ConnectRetryInterval: 30 * time.Second,
ConnectRetry: false,
Store: nil,
OnConnect: nil,
OnConnectionLost: DefaultConnectionLostHandler,
OnConnectAttempt: nil,
WriteTimeout: 0, // 0 represents timeout disabled
MessageChannelDepth: 100,
ResumeSubs: false,
HTTPHeaders: make(map[string][]string),
WebsocketOptions: &WebsocketOptions{},
}
return o
}
@ -149,7 +166,7 @@ func (o *ClientOptions) SetResumeSubs(resume bool) *ClientOptions {
// SetClientID will set the client id to be used by this client when
// connecting to the MQTT broker. According to the MQTT v3.1 specification,
// a client id mus be no longer than 23 characters.
// a client id must be no longer than 23 characters.
func (o *ClientOptions) SetClientID(id string) *ClientOptions {
o.ClientID = id
return o
@ -157,7 +174,7 @@ func (o *ClientOptions) SetClientID(id string) *ClientOptions {
// SetUsername will set the username to be used by this client when connecting
// to the MQTT broker. Note: without the use of SSL/TLS, this information will
// be sent in plaintext accross the wire.
// be sent in plaintext across the wire.
func (o *ClientOptions) SetUsername(u string) *ClientOptions {
o.Username = u
return o
@ -165,7 +182,7 @@ func (o *ClientOptions) SetUsername(u string) *ClientOptions {
// SetPassword will set the password to be used by this client when connecting
// to the MQTT broker. Note: without the use of SSL/TLS, this information will
// be sent in plaintext accross the wire.
// be sent in plaintext across the wire.
func (o *ClientOptions) SetPassword(p string) *ClientOptions {
o.Password = p
return o
@ -174,7 +191,7 @@ func (o *ClientOptions) SetPassword(p string) *ClientOptions {
// SetCredentialsProvider will set a method to be called by this client when
// connecting to the MQTT broker that provide the current username and password.
// Note: without the use of SSL/TLS, this information will be sent
// in plaintext accross the wire.
// in plaintext across the wire.
func (o *ClientOptions) SetCredentialsProvider(p CredentialsProvider) *ClientOptions {
o.CredentialsProvider = p
return o
@ -184,7 +201,7 @@ func (o *ClientOptions) SetCredentialsProvider(p CredentialsProvider) *ClientOpt
// when this client connects to an MQTT broker. By setting this flag, you are
// indicating that no messages saved by the broker for this client should be
// delivered. Any messages that were going to be sent by this client before
// diconnecting previously but didn't will not be sent upon connecting to the
// disconnecting previously but didn't will not be sent upon connecting to the
// broker.
func (o *ClientOptions) SetCleanSession(clean bool) *ClientOptions {
o.CleanSession = clean
@ -192,9 +209,13 @@ func (o *ClientOptions) SetCleanSession(clean bool) *ClientOptions {
}
// SetOrderMatters will set the message routing to guarantee order within
// each QoS level. By default, this value is true. If set to false,
// each QoS level. By default, this value is true. If set to false (recommended),
// this flag indicates that messages can be delivered asynchronously
// from the client to the application and possibly arrive out of order.
// Specifically, the message handler is called in its own go routine.
// Note that setting this to true does not guarantee in-order delivery
// (this is subject to broker settings like "max_inflight_messages=1" in mosquitto)
// and if true then handlers must not block.
func (o *ClientOptions) SetOrderMatters(order bool) *ClientOptions {
o.Order = order
return o
@ -274,6 +295,11 @@ func (o *ClientOptions) SetBinaryWill(topic string, payload []byte, qos byte, re
// SetDefaultPublishHandler sets the MessageHandler that will be called when a message
// is received that does not match any known subscriptions.
//
// If OrderMatters is true (the defaultHandler) then callback must not block or
// call functions within this package that may block (e.g. Publish) other than in
// a new go routine.
// defaultHandler must be safe for concurrent use by multiple goroutines.
func (o *ClientOptions) SetDefaultPublishHandler(defaultHandler MessageHandler) *ClientOptions {
o.DefaultPublishHandler = defaultHandler
return o
@ -293,15 +319,31 @@ func (o *ClientOptions) SetConnectionLostHandler(onLost ConnectionLostHandler) *
return o
}
// SetReconnectingHandler sets the OnReconnecting callback to be executed prior
// to the client attempting a reconnect to the MQTT broker.
func (o *ClientOptions) SetReconnectingHandler(cb ReconnectHandler) *ClientOptions {
o.OnReconnecting = cb
return o
}
// SetConnectionAttemptHandler sets the ConnectionAttemptHandler callback to be executed prior
// to each attempt to connect to an MQTT broker. Returns the *tls.Config that will be used when establishing
// the connection (a copy of the tls.Config from ClientOptions will be passed in along with the broker URL).
// This allows connection specific changes to be made to the *tls.Config.
func (o *ClientOptions) SetConnectionAttemptHandler(onConnectAttempt ConnectionAttemptHandler) *ClientOptions {
o.OnConnectAttempt = onConnectAttempt
return o
}
// SetWriteTimeout puts a limit on how long a mqtt publish should block until it unblocks with a
// timeout error. A duration of 0 never times out. Default 30 seconds
// timeout error. A duration of 0 never times out. Default never times out
func (o *ClientOptions) SetWriteTimeout(t time.Duration) *ClientOptions {
o.WriteTimeout = t
return o
}
// SetConnectTimeout limits how long the client will wait when trying to open a connection
// to an MQTT server before timeing out and erroring the attempt. A duration of 0 never times out.
// to an MQTT server before timing out. A duration of 0 never times out.
// Default 30 seconds. Currently only operational on TCP/TLS connections.
func (o *ClientOptions) SetConnectTimeout(t time.Duration) *ClientOptions {
o.ConnectTimeout = t
@ -323,10 +365,25 @@ func (o *ClientOptions) SetAutoReconnect(a bool) *ClientOptions {
return o
}
// SetMessageChannelDepth sets the size of the internal queue that holds messages while the
// client is temporairily offline, allowing the application to publish when the client is
// reconnecting. This setting is only valid if AutoReconnect is set to true, it is otherwise
// ignored.
// SetConnectRetryInterval sets the time that will be waited between connection attempts
// when initially connecting if ConnectRetry is TRUE
func (o *ClientOptions) SetConnectRetryInterval(t time.Duration) *ClientOptions {
o.ConnectRetryInterval = t
return o
}
// SetConnectRetry sets whether the connect function will automatically retry the connection
// in the event of a failure (when true the token returned by the Connect function will
// not complete until the connection is up or it is cancelled)
// If ConnectRetry is true then subscriptions should be requested in OnConnect handler
// Setting this to TRUE permits messages to be published before the connection is established
func (o *ClientOptions) SetConnectRetry(a bool) *ClientOptions {
o.ConnectRetry = a
return o
}
// SetMessageChannelDepth DEPRECATED The value set here no longer has any effect, this function
// remains so the API is not altered.
func (o *ClientOptions) SetMessageChannelDepth(s uint) *ClientOptions {
o.MessageChannelDepth = s
return o
@ -338,3 +395,9 @@ func (o *ClientOptions) SetHTTPHeaders(h http.Header) *ClientOptions {
o.HTTPHeaders = h
return o
}
// SetWebsocketOptions sets the additional websocket options used in a WebSocket connection
func (o *ClientOptions) SetWebsocketOptions(w *WebsocketOptions) *ClientOptions {
o.WebsocketOptions = w
return o
}

30
vendor/github.com/eclipse/paho.mqtt.golang/options_reader.go generated vendored
View File

@ -26,7 +26,7 @@ type ClientOptionsReader struct {
options *ClientOptions
}
//Servers returns a slice of the servers defined in the clientoptions
// Servers returns a slice of the servers defined in the clientoptions
func (r *ClientOptionsReader) Servers() []*url.URL {
s := make([]*url.URL, len(r.options.Servers))
@ -38,31 +38,31 @@ func (r *ClientOptionsReader) Servers() []*url.URL {
return s
}
//ResumeSubs returns true if resuming stored (un)sub is enabled
// ResumeSubs returns true if resuming stored (un)sub is enabled
func (r *ClientOptionsReader) ResumeSubs() bool {
s := r.options.ResumeSubs
return s
}
//ClientID returns the set client id
// ClientID returns the set client id
func (r *ClientOptionsReader) ClientID() string {
s := r.options.ClientID
return s
}
//Username returns the set username
// Username returns the set username
func (r *ClientOptionsReader) Username() string {
s := r.options.Username
return s
}
//Password returns the set password
// Password returns the set password
func (r *ClientOptionsReader) Password() string {
s := r.options.Password
return s
}
//CleanSession returns whether Cleansession is set
// CleanSession returns whether Cleansession is set
func (r *ClientOptionsReader) CleanSession() bool {
s := r.options.CleanSession
return s
@ -133,6 +133,18 @@ func (r *ClientOptionsReader) AutoReconnect() bool {
return s
}
// ConnectRetryInterval returns the delay between retries on the initial connection (if ConnectRetry true)
func (r *ClientOptionsReader) ConnectRetryInterval() time.Duration {
s := r.options.ConnectRetryInterval
return s
}
// ConnectRetry returns whether the initial connection request will be retried until connection established
func (r *ClientOptionsReader) ConnectRetry() bool {
s := r.options.ConnectRetry
return s
}
func (r *ClientOptionsReader) WriteTimeout() time.Duration {
s := r.options.WriteTimeout
return s
@ -147,3 +159,9 @@ func (r *ClientOptionsReader) HTTPHeaders() http.Header {
h := r.options.HTTPHeaders
return h
}
// WebsocketOptions returns the currently configured WebSocket options
func (r *ClientOptionsReader) WebsocketOptions() *WebsocketOptions {
s := r.options.WebsocketOptions
return s
}

17
vendor/github.com/eclipse/paho.mqtt.golang/packets/connack.go generated vendored
View File

@ -6,8 +6,8 @@ import (
"io"
)
//ConnackPacket is an internal representation of the fields of the
//Connack MQTT packet
// ConnackPacket is an internal representation of the fields of the
// Connack MQTT packet
type ConnackPacket struct {
FixedHeader
SessionPresent bool
@ -15,10 +15,7 @@ type ConnackPacket struct {
}
func (ca *ConnackPacket) String() string {
str := fmt.Sprintf("%s", ca.FixedHeader)
str += " "
str += fmt.Sprintf("sessionpresent: %t returncode: %d", ca.SessionPresent, ca.ReturnCode)
return str
return fmt.Sprintf("%s sessionpresent: %t returncode: %d", ca.FixedHeader, ca.SessionPresent, ca.ReturnCode)
}
func (ca *ConnackPacket) Write(w io.Writer) error {
@ -35,8 +32,8 @@ func (ca *ConnackPacket) Write(w io.Writer) error {
return err
}
//Unpack decodes the details of a ControlPacket after the fixed
//header has been read
// Unpack decodes the details of a ControlPacket after the fixed
// header has been read
func (ca *ConnackPacket) Unpack(b io.Reader) error {
flags, err := decodeByte(b)
if err != nil {
@ -48,8 +45,8 @@ func (ca *ConnackPacket) Unpack(b io.Reader) error {
return err
}
//Details returns a Details struct containing the Qos and
//MessageID of this ControlPacket
// Details returns a Details struct containing the Qos and
// MessageID of this ControlPacket
func (ca *ConnackPacket) Details() Details {
return Details{Qos: 0, MessageID: 0}
}

33
vendor/github.com/eclipse/paho.mqtt.golang/packets/connect.go generated vendored
View File

@ -6,8 +6,8 @@ import (
"io"
)
//ConnectPacket is an internal representation of the fields of the
//Connect MQTT packet
// ConnectPacket is an internal representation of the fields of the
// Connect MQTT packet
type ConnectPacket struct {
FixedHeader
ProtocolName string
@ -29,10 +29,11 @@ type ConnectPacket struct {
}
func (c *ConnectPacket) String() string {
str := fmt.Sprintf("%s", c.FixedHeader)
str += " "
str += fmt.Sprintf("protocolversion: %d protocolname: %s cleansession: %t willflag: %t WillQos: %d WillRetain: %t Usernameflag: %t Passwordflag: %t keepalive: %d clientId: %s willtopic: %s willmessage: %s Username: %s Password: %s", c.ProtocolVersion, c.ProtocolName, c.CleanSession, c.WillFlag, c.WillQos, c.WillRetain, c.UsernameFlag, c.PasswordFlag, c.Keepalive, c.ClientIdentifier, c.WillTopic, c.WillMessage, c.Username, c.Password)
return str
var password string
if len(c.Password) > 0 {
password = "<redacted>"
}
return fmt.Sprintf("%s protocolversion: %d protocolname: %s cleansession: %t willflag: %t WillQos: %d WillRetain: %t Usernameflag: %t Passwordflag: %t keepalive: %d clientId: %s willtopic: %s willmessage: %s Username: %s Password: %s", c.FixedHeader, c.ProtocolVersion, c.ProtocolName, c.CleanSession, c.WillFlag, c.WillQos, c.WillRetain, c.UsernameFlag, c.PasswordFlag, c.Keepalive, c.ClientIdentifier, c.WillTopic, c.WillMessage, c.Username, password)
}
func (c *ConnectPacket) Write(w io.Writer) error {
@ -62,8 +63,8 @@ func (c *ConnectPacket) Write(w io.Writer) error {
return err
}
//Unpack decodes the details of a ControlPacket after the fixed
//header has been read
// Unpack decodes the details of a ControlPacket after the fixed
// header has been read
func (c *ConnectPacket) Unpack(b io.Reader) error {
var err error
c.ProtocolName, err = decodeString(b)
@ -119,36 +120,36 @@ func (c *ConnectPacket) Unpack(b io.Reader) error {
return nil
}
//Validate performs validation of the fields of a Connect packet
// Validate performs validation of the fields of a Connect packet
func (c *ConnectPacket) Validate() byte {
if c.PasswordFlag && !c.UsernameFlag {
return ErrRefusedBadUsernameOrPassword
}
if c.ReservedBit != 0 {
//Bad reserved bit
// Bad reserved bit
return ErrProtocolViolation
}
if (c.ProtocolName == "MQIsdp" && c.ProtocolVersion != 3) || (c.ProtocolName == "MQTT" && c.ProtocolVersion != 4) {
//Mismatched or unsupported protocol version
// Mismatched or unsupported protocol version
return ErrRefusedBadProtocolVersion
}
if c.ProtocolName != "MQIsdp" && c.ProtocolName != "MQTT" {
//Bad protocol name
// Bad protocol name
return ErrProtocolViolation
}
if len(c.ClientIdentifier) > 65535 || len(c.Username) > 65535 || len(c.Password) > 65535 {
//Bad size field
// Bad size field
return ErrProtocolViolation
}
if len(c.ClientIdentifier) == 0 && !c.CleanSession {
//Bad client identifier
// Bad client identifier
return ErrRefusedIDRejected
}
return Accepted
}
//Details returns a Details struct containing the Qos and
//MessageID of this ControlPacket
// Details returns a Details struct containing the Qos and
// MessageID of this ControlPacket
func (c *ConnectPacket) Details() Details {
return Details{Qos: 0, MessageID: 0}
}

16
vendor/github.com/eclipse/paho.mqtt.golang/packets/disconnect.go generated vendored
View File

@ -1,19 +1,17 @@
package packets
import (
"fmt"
"io"
)
//DisconnectPacket is an internal representation of the fields of the
//Disconnect MQTT packet
// DisconnectPacket is an internal representation of the fields of the
// Disconnect MQTT packet
type DisconnectPacket struct {
FixedHeader
}
func (d *DisconnectPacket) String() string {
str := fmt.Sprintf("%s", d.FixedHeader)
return str
return d.FixedHeader.String()
}
func (d *DisconnectPacket) Write(w io.Writer) error {
@ -23,14 +21,14 @@ func (d *DisconnectPacket) Write(w io.Writer) error {
return err
}
//Unpack decodes the details of a ControlPacket after the fixed
//header has been read
// Unpack decodes the details of a ControlPacket after the fixed
// header has been read
func (d *DisconnectPacket) Unpack(b io.Reader) error {
return nil
}
//Details returns a Details struct containing the Qos and
//MessageID of this ControlPacket
// Details returns a Details struct containing the Qos and
// MessageID of this ControlPacket
func (d *DisconnectPacket) Details() Details {
return Details{Qos: 0, MessageID: 0}
}

90
vendor/github.com/eclipse/paho.mqtt.golang/packets/packets.go generated vendored
View File

@ -8,9 +8,9 @@ import (
"io"
)
//ControlPacket defines the interface for structs intended to hold
//decoded MQTT packets, either from being read or before being
//written
// ControlPacket defines the interface for structs intended to hold
// decoded MQTT packets, either from being read or before being
// written
type ControlPacket interface {
Write(io.Writer) error
Unpack(io.Reader) error
@ -18,8 +18,8 @@ type ControlPacket interface {
Details() Details
}
//PacketNames maps the constants for each of the MQTT packet types
//to a string representation of their name.
// PacketNames maps the constants for each of the MQTT packet types
// to a string representation of their name.
var PacketNames = map[uint8]string{
1: "CONNECT",
2: "CONNACK",
@ -37,7 +37,7 @@ var PacketNames = map[uint8]string{
14: "DISCONNECT",
}
//Below are the constants assigned to each of the MQTT packet types
// Below are the constants assigned to each of the MQTT packet types
const (
Connect = 1
Connack = 2
@ -55,8 +55,8 @@ const (
Disconnect = 14
)
//Below are the const definitions for error codes returned by
//Connect()
// Below are the const definitions for error codes returned by
// Connect()
const (
Accepted = 0x00
ErrRefusedBadProtocolVersion = 0x01
@ -68,8 +68,8 @@ const (
ErrProtocolViolation = 0xFF
)
//ConnackReturnCodes is a map of the error codes constants for Connect()
//to a string representation of the error
// ConnackReturnCodes is a map of the error codes constants for Connect()
// to a string representation of the error
var ConnackReturnCodes = map[uint8]string{
0: "Connection Accepted",
1: "Connection Refused: Bad Protocol Version",
@ -81,23 +81,33 @@ var ConnackReturnCodes = map[uint8]string{
255: "Connection Refused: Protocol Violation",
}
//ConnErrors is a map of the errors codes constants for Connect()
//to a Go error
var (
ErrorRefusedBadProtocolVersion = errors.New("unacceptable protocol version")
ErrorRefusedIDRejected = errors.New("identifier rejected")
ErrorRefusedServerUnavailable = errors.New("server Unavailable")
ErrorRefusedBadUsernameOrPassword = errors.New("bad user name or password")
ErrorRefusedNotAuthorised = errors.New("not Authorized")
ErrorNetworkError = errors.New("network Error")
ErrorProtocolViolation = errors.New("protocol Violation")
)
// ConnErrors is a map of the errors codes constants for Connect()
// to a Go error
var ConnErrors = map[byte]error{
Accepted: nil,
ErrRefusedBadProtocolVersion: errors.New("Unnacceptable protocol version"),
ErrRefusedIDRejected: errors.New("Identifier rejected"),
ErrRefusedServerUnavailable: errors.New("Server Unavailable"),
ErrRefusedBadUsernameOrPassword: errors.New("Bad user name or password"),
ErrRefusedNotAuthorised: errors.New("Not Authorized"),
ErrNetworkError: errors.New("Network Error"),
ErrProtocolViolation: errors.New("Protocol Violation"),
ErrRefusedBadProtocolVersion: ErrorRefusedBadProtocolVersion,
ErrRefusedIDRejected: ErrorRefusedIDRejected,
ErrRefusedServerUnavailable: ErrorRefusedServerUnavailable,
ErrRefusedBadUsernameOrPassword: ErrorRefusedBadUsernameOrPassword,
ErrRefusedNotAuthorised: ErrorRefusedNotAuthorised,
ErrNetworkError: ErrorNetworkError,
ErrProtocolViolation: ErrorProtocolViolation,
}
//ReadPacket takes an instance of an io.Reader (such as net.Conn) and attempts
//to read an MQTT packet from the stream. It returns a ControlPacket
//representing the decoded MQTT packet and an error. One of these returns will
//always be nil, a nil ControlPacket indicating an error occurred.
// ReadPacket takes an instance of an io.Reader (such as net.Conn) and attempts
// to read an MQTT packet from the stream. It returns a ControlPacket
// representing the decoded MQTT packet and an error. One of these returns will
// always be nil, a nil ControlPacket indicating an error occurred.
func ReadPacket(r io.Reader) (ControlPacket, error) {
var fh FixedHeader
b := make([]byte, 1)
@ -123,17 +133,17 @@ func ReadPacket(r io.Reader) (ControlPacket, error) {
return nil, err
}
if n != fh.RemainingLength {
return nil, errors.New("Failed to read expected data")
return nil, errors.New("failed to read expected data")
}
err = cp.Unpack(bytes.NewBuffer(packetBytes))
return cp, err
}
//NewControlPacket is used to create a new ControlPacket of the type specified
//by packetType, this is usually done by reference to the packet type constants
//defined in packets.go. The newly created ControlPacket is empty and a pointer
//is returned.
// NewControlPacket is used to create a new ControlPacket of the type specified
// by packetType, this is usually done by reference to the packet type constants
// defined in packets.go. The newly created ControlPacket is empty and a pointer
// is returned.
func NewControlPacket(packetType byte) ControlPacket {
switch packetType {
case Connect:
@ -168,9 +178,9 @@ func NewControlPacket(packetType byte) ControlPacket {
return nil
}
//NewControlPacketWithHeader is used to create a new ControlPacket of the type
//specified within the FixedHeader that is passed to the function.
//The newly created ControlPacket is empty and a pointer is returned.
// NewControlPacketWithHeader is used to create a new ControlPacket of the type
// specified within the FixedHeader that is passed to the function.
// The newly created ControlPacket is empty and a pointer is returned.
func NewControlPacketWithHeader(fh FixedHeader) (ControlPacket, error) {
switch fh.MessageType {
case Connect:
@ -205,16 +215,16 @@ func NewControlPacketWithHeader(fh FixedHeader) (ControlPacket, error) {
return nil, fmt.Errorf("unsupported packet type 0x%x", fh.MessageType)
}
//Details struct returned by the Details() function called on
//ControlPackets to present details of the Qos and MessageID
//of the ControlPacket
// Details struct returned by the Details() function called on
// ControlPackets to present details of the Qos and MessageID
// of the ControlPacket
type Details struct {
Qos byte
MessageID uint16
}
//FixedHeader is a struct to hold the decoded information from
//the fixed header of an MQTT ControlPacket
// FixedHeader is a struct to hold the decoded information from
// the fixed header of an MQTT ControlPacket
type FixedHeader struct {
MessageType byte
Dup bool
@ -274,9 +284,9 @@ func decodeUint16(b io.Reader) (uint16, error) {
}
func encodeUint16(num uint16) []byte {
bytes := make([]byte, 2)
binary.BigEndian.PutUint16(bytes, num)
return bytes
bytesResult := make([]byte, 2)
binary.BigEndian.PutUint16(bytesResult, num)
return bytesResult
}
func encodeString(field string) []byte {
@ -329,7 +339,7 @@ func decodeLength(r io.Reader) (int, error) {
var rLength uint32
var multiplier uint32
b := make([]byte, 1)
for multiplier < 27 { //fix: Infinite '(digit & 128) == 1' will cause the dead loop
for multiplier < 27 { // fix: Infinite '(digit & 128) == 1' will cause the dead loop
_, err := io.ReadFull(r, b)
if err != nil {
return 0, err

16
vendor/github.com/eclipse/paho.mqtt.golang/packets/pingreq.go generated vendored
View File

@ -1,19 +1,17 @@
package packets
import (
"fmt"
"io"
)
//PingreqPacket is an internal representation of the fields of the
//Pingreq MQTT packet
// PingreqPacket is an internal representation of the fields of the
// Pingreq MQTT packet
type PingreqPacket struct {
FixedHeader
}
func (pr *PingreqPacket) String() string {
str := fmt.Sprintf("%s", pr.FixedHeader)
return str
return pr.FixedHeader.String()
}
func (pr *PingreqPacket) Write(w io.Writer) error {
@ -23,14 +21,14 @@ func (pr *PingreqPacket) Write(w io.Writer) error {
return err
}
//Unpack decodes the details of a ControlPacket after the fixed
//header has been read
// Unpack decodes the details of a ControlPacket after the fixed
// header has been read
func (pr *PingreqPacket) Unpack(b io.Reader) error {
return nil
}
//Details returns a Details struct containing the Qos and
//MessageID of this ControlPacket
// Details returns a Details struct containing the Qos and
// MessageID of this ControlPacket
func (pr *PingreqPacket) Details() Details {
return Details{Qos: 0, MessageID: 0}
}

16
vendor/github.com/eclipse/paho.mqtt.golang/packets/pingresp.go generated vendored
View File

@ -1,19 +1,17 @@
package packets
import (
"fmt"
"io"
)
//PingrespPacket is an internal representation of the fields of the
//Pingresp MQTT packet
// PingrespPacket is an internal representation of the fields of the
// Pingresp MQTT packet
type PingrespPacket struct {
FixedHeader
}
func (pr *PingrespPacket) String() string {
str := fmt.Sprintf("%s", pr.FixedHeader)
return str
return pr.FixedHeader.String()
}
func (pr *PingrespPacket) Write(w io.Writer) error {
@ -23,14 +21,14 @@ func (pr *PingrespPacket) Write(w io.Writer) error {
return err
}
//Unpack decodes the details of a ControlPacket after the fixed
//header has been read
// Unpack decodes the details of a ControlPacket after the fixed
// header has been read
func (pr *PingrespPacket) Unpack(b io.Reader) error {
return nil
}
//Details returns a Details struct containing the Qos and
//MessageID of this ControlPacket
// Details returns a Details struct containing the Qos and
// MessageID of this ControlPacket
func (pr *PingrespPacket) Details() Details {
return Details{Qos: 0, MessageID: 0}
}

17
vendor/github.com/eclipse/paho.mqtt.golang/packets/puback.go generated vendored
View File

@ -5,18 +5,15 @@ import (
"io"
)
//PubackPacket is an internal representation of the fields of the
//Puback MQTT packet
// PubackPacket is an internal representation of the fields of the
// Puback MQTT packet
type PubackPacket struct {
FixedHeader
MessageID uint16
}
func (pa *PubackPacket) String() string {
str := fmt.Sprintf("%s", pa.FixedHeader)
str += " "
str += fmt.Sprintf("MessageID: %d", pa.MessageID)
return str
return fmt.Sprintf("%s MessageID: %d", pa.FixedHeader, pa.MessageID)
}
func (pa *PubackPacket) Write(w io.Writer) error {
@ -29,8 +26,8 @@ func (pa *PubackPacket) Write(w io.Writer) error {
return err
}
//Unpack decodes the details of a ControlPacket after the fixed
//header has been read
// Unpack decodes the details of a ControlPacket after the fixed
// header has been read
func (pa *PubackPacket) Unpack(b io.Reader) error {
var err error
pa.MessageID, err = decodeUint16(b)
@ -38,8 +35,8 @@ func (pa *PubackPacket) Unpack(b io.Reader) error {
return err
}
//Details returns a Details struct containing the Qos and
//MessageID of this ControlPacket
// Details returns a Details struct containing the Qos and
// MessageID of this ControlPacket
func (pa *PubackPacket) Details() Details {
return Details{Qos: pa.Qos, MessageID: pa.MessageID}
}

17
vendor/github.com/eclipse/paho.mqtt.golang/packets/pubcomp.go generated vendored
View File

@ -5,18 +5,15 @@ import (
"io"
)
//PubcompPacket is an internal representation of the fields of the
//Pubcomp MQTT packet
// PubcompPacket is an internal representation of the fields of the
// Pubcomp MQTT packet
type PubcompPacket struct {
FixedHeader
MessageID uint16
}
func (pc *PubcompPacket) String() string {
str := fmt.Sprintf("%s", pc.FixedHeader)
str += " "
str += fmt.Sprintf("MessageID: %d", pc.MessageID)
return str
return fmt.Sprintf("%s MessageID: %d", pc.FixedHeader, pc.MessageID)
}
func (pc *PubcompPacket) Write(w io.Writer) error {
@ -29,8 +26,8 @@ func (pc *PubcompPacket) Write(w io.Writer) error {
return err
}
//Unpack decodes the details of a ControlPacket after the fixed
//header has been read
// Unpack decodes the details of a ControlPacket after the fixed
// header has been read
func (pc *PubcompPacket) Unpack(b io.Reader) error {
var err error
pc.MessageID, err = decodeUint16(b)
@ -38,8 +35,8 @@ func (pc *PubcompPacket) Unpack(b io.Reader) error {
return err
}
//Details returns a Details struct containing the Qos and
//MessageID of this ControlPacket
// Details returns a Details struct containing the Qos and
// MessageID of this ControlPacket
func (pc *PubcompPacket) Details() Details {
return Details{Qos: pc.Qos, MessageID: pc.MessageID}
}

29
vendor/github.com/eclipse/paho.mqtt.golang/packets/publish.go generated vendored
View File

@ -6,8 +6,8 @@ import (
"io"
)
//PublishPacket is an internal representation of the fields of the
//Publish MQTT packet
// PublishPacket is an internal representation of the fields of the
// Publish MQTT packet
type PublishPacket struct {
FixedHeader
TopicName string
@ -16,12 +16,7 @@ type PublishPacket struct {
}
func (p *PublishPacket) String() string {
str := fmt.Sprintf("%s", p.FixedHeader)
str += " "
str += fmt.Sprintf("topicName: %s MessageID: %d", p.TopicName, p.MessageID)
str += " "
str += fmt.Sprintf("payload: %s", string(p.Payload))
return str
return fmt.Sprintf("%s topicName: %s MessageID: %d payload: %s", p.FixedHeader, p.TopicName, p.MessageID, string(p.Payload))
}
func (p *PublishPacket) Write(w io.Writer) error {
@ -41,8 +36,8 @@ func (p *PublishPacket) Write(w io.Writer) error {
return err
}
//Unpack decodes the details of a ControlPacket after the fixed
//header has been read
// Unpack decodes the details of a ControlPacket after the fixed
// header has been read
func (p *PublishPacket) Unpack(b io.Reader) error {
var payloadLength = p.FixedHeader.RemainingLength
var err error
@ -61,7 +56,7 @@ func (p *PublishPacket) Unpack(b io.Reader) error {
payloadLength -= len(p.TopicName) + 2
}
if payloadLength < 0 {
return fmt.Errorf("Error unpacking publish, payload length < 0")
return fmt.Errorf("error unpacking publish, payload length < 0")
}
p.Payload = make([]byte, payloadLength)
_, err = b.Read(p.Payload)
@ -69,10 +64,10 @@ func (p *PublishPacket) Unpack(b io.Reader) error {
return err
}
//Copy creates a new PublishPacket with the same topic and payload
//but an empty fixed header, useful for when you want to deliver
//a message with different properties such as Qos but the same
//content
// Copy creates a new PublishPacket with the same topic and payload
// but an empty fixed header, useful for when you want to deliver
// a message with different properties such as Qos but the same
// content
func (p *PublishPacket) Copy() *PublishPacket {
newP := NewControlPacket(Publish).(*PublishPacket)
newP.TopicName = p.TopicName
@ -81,8 +76,8 @@ func (p *PublishPacket) Copy() *PublishPacket {
return newP
}
//Details returns a Details struct containing the Qos and
//MessageID of this ControlPacket
// Details returns a Details struct containing the Qos and
// MessageID of this ControlPacket
func (p *PublishPacket) Details() Details {
return Details{Qos: p.Qos, MessageID: p.MessageID}
}

17
vendor/github.com/eclipse/paho.mqtt.golang/packets/pubrec.go generated vendored
View File

@ -5,18 +5,15 @@ import (
"io"
)
//PubrecPacket is an internal representation of the fields of the
//Pubrec MQTT packet
// PubrecPacket is an internal representation of the fields of the
// Pubrec MQTT packet
type PubrecPacket struct {
FixedHeader
MessageID uint16
}
func (pr *PubrecPacket) String() string {
str := fmt.Sprintf("%s", pr.FixedHeader)
str += " "
str += fmt.Sprintf("MessageID: %d", pr.MessageID)
return str
return fmt.Sprintf("%s MessageID: %d", pr.FixedHeader, pr.MessageID)
}
func (pr *PubrecPacket) Write(w io.Writer) error {
@ -29,8 +26,8 @@ func (pr *PubrecPacket) Write(w io.Writer) error {
return err
}
//Unpack decodes the details of a ControlPacket after the fixed
//header has been read
// Unpack decodes the details of a ControlPacket after the fixed
// header has been read
func (pr *PubrecPacket) Unpack(b io.Reader) error {
var err error
pr.MessageID, err = decodeUint16(b)
@ -38,8 +35,8 @@ func (pr *PubrecPacket) Unpack(b io.Reader) error {
return err
}
//Details returns a Details struct containing the Qos and
//MessageID of this ControlPacket
// Details returns a Details struct containing the Qos and
// MessageID of this ControlPacket
func (pr *PubrecPacket) Details() Details {
return Details{Qos: pr.Qos, MessageID: pr.MessageID}
}

17
vendor/github.com/eclipse/paho.mqtt.golang/packets/pubrel.go generated vendored
View File

@ -5,18 +5,15 @@ import (
"io"
)
//PubrelPacket is an internal representation of the fields of the
//Pubrel MQTT packet
// PubrelPacket is an internal representation of the fields of the
// Pubrel MQTT packet
type PubrelPacket struct {
FixedHeader
MessageID uint16
}
func (pr *PubrelPacket) String() string {
str := fmt.Sprintf("%s", pr.FixedHeader)
str += " "
str += fmt.Sprintf("MessageID: %d", pr.MessageID)
return str
return fmt.Sprintf("%s MessageID: %d", pr.FixedHeader, pr.MessageID)
}
func (pr *PubrelPacket) Write(w io.Writer) error {
@ -29,8 +26,8 @@ func (pr *PubrelPacket) Write(w io.Writer) error {
return err
}
//Unpack decodes the details of a ControlPacket after the fixed
//header has been read
// Unpack decodes the details of a ControlPacket after the fixed
// header has been read
func (pr *PubrelPacket) Unpack(b io.Reader) error {
var err error
pr.MessageID, err = decodeUint16(b)
@ -38,8 +35,8 @@ func (pr *PubrelPacket) Unpack(b io.Reader) error {
return err
}
//Details returns a Details struct containing the Qos and
//MessageID of this ControlPacket
// Details returns a Details struct containing the Qos and
// MessageID of this ControlPacket
func (pr *PubrelPacket) Details() Details {
return Details{Qos: pr.Qos, MessageID: pr.MessageID}
}

17
vendor/github.com/eclipse/paho.mqtt.golang/packets/suback.go generated vendored
View File

@ -6,8 +6,8 @@ import (
"io"
)
//SubackPacket is an internal representation of the fields of the
//Suback MQTT packet
// SubackPacket is an internal representation of the fields of the
// Suback MQTT packet
type SubackPacket struct {
FixedHeader
MessageID uint16
@ -15,10 +15,7 @@ type SubackPacket struct {
}
func (sa *SubackPacket) String() string {
str := fmt.Sprintf("%s", sa.FixedHeader)
str += " "
str += fmt.Sprintf("MessageID: %d", sa.MessageID)
return str
return fmt.Sprintf("%s MessageID: %d", sa.FixedHeader, sa.MessageID)
}
func (sa *SubackPacket) Write(w io.Writer) error {
@ -34,8 +31,8 @@ func (sa *SubackPacket) Write(w io.Writer) error {
return err
}
//Unpack decodes the details of a ControlPacket after the fixed
//header has been read
// Unpack decodes the details of a ControlPacket after the fixed
// header has been read
func (sa *SubackPacket) Unpack(b io.Reader) error {
var qosBuffer bytes.Buffer
var err error
@ -53,8 +50,8 @@ func (sa *SubackPacket) Unpack(b io.Reader) error {
return nil
}
//Details returns a Details struct containing the Qos and
//MessageID of this ControlPacket
// Details returns a Details struct containing the Qos and
// MessageID of this ControlPacket
func (sa *SubackPacket) Details() Details {
return Details{Qos: 0, MessageID: sa.MessageID}
}

19
vendor/github.com/eclipse/paho.mqtt.golang/packets/subscribe.go generated vendored
View File

@ -6,8 +6,8 @@ import (
"io"
)
//SubscribePacket is an internal representation of the fields of the
//Subscribe MQTT packet
// SubscribePacket is an internal representation of the fields of the
// Subscribe MQTT packet
type SubscribePacket struct {
FixedHeader
MessageID uint16
@ -16,10 +16,7 @@ type SubscribePacket struct {
}
func (s *SubscribePacket) String() string {
str := fmt.Sprintf("%s", s.FixedHeader)
str += " "
str += fmt.Sprintf("MessageID: %d topics: %s", s.MessageID, s.Topics)
return str
return fmt.Sprintf("%s MessageID: %d topics: %s", s.FixedHeader, s.MessageID, s.Topics)
}
func (s *SubscribePacket) Write(w io.Writer) error {
@ -39,8 +36,8 @@ func (s *SubscribePacket) Write(w io.Writer) error {
return err
}
//Unpack decodes the details of a ControlPacket after the fixed
//header has been read
// Unpack decodes the details of a ControlPacket after the fixed
// header has been read
func (s *SubscribePacket) Unpack(b io.Reader) error {
var err error
s.MessageID, err = decodeUint16(b)
@ -59,14 +56,14 @@ func (s *SubscribePacket) Unpack(b io.Reader) error {
return err
}
s.Qoss = append(s.Qoss, qos)
payloadLength -= 2 + len(topic) + 1 //2 bytes of string length, plus string, plus 1 byte for Qos
payloadLength -= 2 + len(topic) + 1 // 2 bytes of string length, plus string, plus 1 byte for Qos
}
return nil
}
//Details returns a Details struct containing the Qos and
//MessageID of this ControlPacket
// Details returns a Details struct containing the Qos and
// MessageID of this ControlPacket
func (s *SubscribePacket) Details() Details {
return Details{Qos: 1, MessageID: s.MessageID}
}

17
vendor/github.com/eclipse/paho.mqtt.golang/packets/unsuback.go generated vendored
View File

@ -5,18 +5,15 @@ import (
"io"
)
//UnsubackPacket is an internal representation of the fields of the
//Unsuback MQTT packet
// UnsubackPacket is an internal representation of the fields of the
// Unsuback MQTT packet
type UnsubackPacket struct {
FixedHeader
MessageID uint16
}
func (ua *UnsubackPacket) String() string {
str := fmt.Sprintf("%s", ua.FixedHeader)
str += " "
str += fmt.Sprintf("MessageID: %d", ua.MessageID)
return str
return fmt.Sprintf("%s MessageID: %d", ua.FixedHeader, ua.MessageID)
}
func (ua *UnsubackPacket) Write(w io.Writer) error {
@ -29,8 +26,8 @@ func (ua *UnsubackPacket) Write(w io.Writer) error {
return err
}
//Unpack decodes the details of a ControlPacket after the fixed
//header has been read
// Unpack decodes the details of a ControlPacket after the fixed
// header has been read
func (ua *UnsubackPacket) Unpack(b io.Reader) error {
var err error
ua.MessageID, err = decodeUint16(b)
@ -38,8 +35,8 @@ func (ua *UnsubackPacket) Unpack(b io.Reader) error {
return err
}
//Details returns a Details struct containing the Qos and
//MessageID of this ControlPacket
// Details returns a Details struct containing the Qos and
// MessageID of this ControlPacket
func (ua *UnsubackPacket) Details() Details {
return Details{Qos: 0, MessageID: ua.MessageID}
}

17
vendor/github.com/eclipse/paho.mqtt.golang/packets/unsubscribe.go generated vendored
View File

@ -6,8 +6,8 @@ import (
"io"
)
//UnsubscribePacket is an internal representation of the fields of the
//Unsubscribe MQTT packet
// UnsubscribePacket is an internal representation of the fields of the
// Unsubscribe MQTT packet
type UnsubscribePacket struct {
FixedHeader
MessageID uint16
@ -15,10 +15,7 @@ type UnsubscribePacket struct {
}
func (u *UnsubscribePacket) String() string {
str := fmt.Sprintf("%s", u.FixedHeader)
str += " "
str += fmt.Sprintf("MessageID: %d", u.MessageID)
return str
return fmt.Sprintf("%s MessageID: %d", u.FixedHeader, u.MessageID)
}
func (u *UnsubscribePacket) Write(w io.Writer) error {
@ -36,8 +33,8 @@ func (u *UnsubscribePacket) Write(w io.Writer) error {
return err
}
//Unpack decodes the details of a ControlPacket after the fixed
//header has been read
// Unpack decodes the details of a ControlPacket after the fixed
// header has been read
func (u *UnsubscribePacket) Unpack(b io.Reader) error {
var err error
u.MessageID, err = decodeUint16(b)
@ -52,8 +49,8 @@ func (u *UnsubscribePacket) Unpack(b io.Reader) error {
return err
}
//Details returns a Details struct containing the Qos and
//MessageID of this ControlPacket
// Details returns a Details struct containing the Qos and
// MessageID of this ControlPacket
func (u *UnsubscribePacket) Details() Details {
return Details{Qos: 1, MessageID: u.MessageID}
}

17
vendor/github.com/eclipse/paho.mqtt.golang/ping.go generated vendored
View File

@ -16,13 +16,16 @@ package mqtt
import (
"errors"
"io"
"sync/atomic"
"time"
"github.com/eclipse/paho.mqtt.golang/packets"
)
func keepalive(c *client) {
// keepalive - Send ping when connection unused for set period
// connection passed in to avoid race condition on shutdown
func keepalive(c *client, conn io.Writer) {
defer c.workers.Done()
DEBUG.Println(PNG, "keepalive starting")
var checkInterval int64
@ -51,17 +54,19 @@ func keepalive(c *client) {
if atomic.LoadInt32(&c.pingOutstanding) == 0 {
DEBUG.Println(PNG, "keepalive sending ping")
ping := packets.NewControlPacket(packets.Pingreq).(*packets.PingreqPacket)
//We don't want to wait behind large messages being sent, the Write call
//will block until it it able to send the packet.
// We don't want to wait behind large messages being sent, the Write call
// will block until it it able to send the packet.
atomic.StoreInt32(&c.pingOutstanding, 1)
ping.Write(c.conn)
if err := ping.Write(conn); err != nil {
ERROR.Println(PNG, err)
}
c.lastSent.Store(time.Now())
pingSent = time.Now()
}
}
if atomic.LoadInt32(&c.pingOutstanding) > 0 && time.Now().Sub(pingSent) >= c.options.PingTimeout {
if atomic.LoadInt32(&c.pingOutstanding) > 0 && time.Since(pingSent) >= c.options.PingTimeout {
CRITICAL.Println(PNG, "pingresp not received, disconnecting")
c.errors <- errors.New("pingresp not received, disconnecting")
c.internalConnLost(errors.New("pingresp not received, disconnecting")) // no harm in calling this if the connection is already down (or shutdown is in progress)
return
}
}

132
vendor/github.com/eclipse/paho.mqtt.golang/router.go generated vendored
View File

@ -37,17 +37,11 @@ type route struct {
// and returns a boolean of the outcome
func match(route []string, topic []string) bool {
if len(route) == 0 {
if len(topic) == 0 {
return true
}
return false
return len(topic) == 0
}
if len(topic) == 0 {
if route[0] == "#" {
return true
}
return false
return route[0] == "#"
}
if route[0] == "#" {
@ -87,15 +81,13 @@ type router struct {
routes *list.List
defaultHandler MessageHandler
messages chan *packets.PublishPacket
stop chan bool
}
// newRouter returns a new instance of a Router and channel which can be used to tell the Router
// to stop
func newRouter() (*router, chan bool) {
router := &router{routes: list.New(), messages: make(chan *packets.PublishPacket), stop: make(chan bool)}
stop := router.stop
return router, stop
func newRouter() *router {
router := &router{routes: list.New(), messages: make(chan *packets.PublishPacket)}
return router
}
// addRoute takes a topic string and MessageHandler callback. It looks in the current list of
@ -105,7 +97,7 @@ func (r *router) addRoute(topic string, callback MessageHandler) {
r.Lock()
defer r.Unlock()
for e := r.routes.Front(); e != nil; e = e.Next() {
if e.Value.(*route).match(topic) {
if e.Value.(*route).topic == topic {
r := e.Value.(*route)
r.callback = callback
return
@ -120,7 +112,7 @@ func (r *router) deleteRoute(topic string) {
r.Lock()
defer r.Unlock()
for e := r.routes.Front(); e != nil; e = e.Next() {
if e.Value.(*route).match(topic) {
if e.Value.(*route).topic == topic {
r.routes.Remove(e)
return
}
@ -139,49 +131,99 @@ func (r *router) setDefaultHandler(handler MessageHandler) {
// takes messages off the channel, matches them against the internal route list and calls the
// associated callback (or the defaultHandler, if one exists and no other route matched). If
// anything is sent down the stop channel the function will end.
func (r *router) matchAndDispatch(messages <-chan *packets.PublishPacket, order bool, client *client) {
go func() {
for {
select {
case message := <-messages:
sent := false
r.RLock()
m := messageFromPublish(message, client.ackFunc(message))
handlers := []MessageHandler{}
for e := r.routes.Front(); e != nil; e = e.Next() {
if e.Value.(*route).match(message.TopicName) {
if order {
handlers = append(handlers, e.Value.(*route).callback)
} else {
hd := e.Value.(*route).callback
go func() {
hd(client, m)
m.Ack()
}()
func (r *router) matchAndDispatch(messages <-chan *packets.PublishPacket, order bool, client *client) <-chan *PacketAndToken {
var wg sync.WaitGroup
ackOutChan := make(chan *PacketAndToken) // Channel returned to caller; closed when messages channel closed
var ackInChan chan *PacketAndToken // ACKs generated by ackFunc get put onto this channel
stopAckCopy := make(chan struct{}) // Closure requests stop of go routine copying ackInChan to ackOutChan
ackCopyStopped := make(chan struct{}) // Closure indicates that it is safe to close ackOutChan
goRoutinesDone := make(chan struct{}) // closed on wg.Done()
if order {
ackInChan = ackOutChan // When order = true no go routines are used so safe to use one channel and close when done
} else {
// When order = false ACK messages are sent in go routines so ackInChan cannot be closed until all goroutines done
ackInChan = make(chan *PacketAndToken)
go func() { // go routine to copy from ackInChan to ackOutChan until stopped
for {
select {
case a := <-ackInChan:
ackOutChan <- a
case <-stopAckCopy:
close(ackCopyStopped) // Signal main go routine that it is safe to close ackOutChan
for {
select {
case <-ackInChan: // drain ackInChan to ensure all goRoutines can complete cleanly (ACK dropped)
DEBUG.Println(ROU, "matchAndDispatch received acknowledgment after processing stopped (ACK dropped).")
case <-goRoutinesDone:
close(ackInChan) // Nothing further should be sent (a panic is probably better than silent failure)
DEBUG.Println(ROU, "matchAndDispatch order=false copy goroutine exiting.")
return
}
sent = true
}
}
if !sent && r.defaultHandler != nil {
}
}()
}
go func() { // Main go routine handling inbound messages
for message := range messages {
// DEBUG.Println(ROU, "matchAndDispatch received message")
sent := false
r.RLock()
m := messageFromPublish(message, ackFunc(ackInChan, client.persist, message))
var handlers []MessageHandler
for e := r.routes.Front(); e != nil; e = e.Next() {
if e.Value.(*route).match(message.TopicName) {
if order {
handlers = append(handlers, e.Value.(*route).callback)
} else {
hd := e.Value.(*route).callback
wg.Add(1)
go func() {
hd(client, m)
m.Ack()
wg.Done()
}()
}
sent = true
}
}
if !sent {
if r.defaultHandler != nil {
if order {
handlers = append(handlers, r.defaultHandler)
} else {
wg.Add(1)
go func() {
r.defaultHandler(client, m)
m.Ack()
wg.Done()
}()
}
} else {
DEBUG.Println(ROU, "matchAndDispatch received message and no handler was available. Message will NOT be acknowledged.")
}
r.RUnlock()
for _, handler := range handlers {
func() {
handler(client, m)
m.Ack()
}()
}
case <-r.stop:
return
}
r.RUnlock()
for _, handler := range handlers {
handler(client, m)
m.Ack()
}
// DEBUG.Println(ROU, "matchAndDispatch handled message")
}
if order {
close(ackOutChan)
} else { // Ensure that nothing further will be written to ackOutChan before closing it
close(stopAckCopy)
<-ackCopyStopped
close(ackOutChan)
go func() {
wg.Wait() // Note: If this remains running then the user has handlers that are not returning
close(goRoutinesDone)
}()
}
DEBUG.Println(ROU, "matchAndDispatch exiting")
}()
return ackOutChan
}

36
vendor/github.com/eclipse/paho.mqtt.golang/token.go generated vendored
View File

@ -31,8 +31,24 @@ type PacketAndToken struct {
// Token defines the interface for the tokens used to indicate when
// actions have completed.
type Token interface {
// Wait will wait indefinitely for the Token to complete, ie the Publish
// to be sent and confirmed receipt from the broker.
Wait() bool
// WaitTimeout takes a time.Duration to wait for the flow associated with the
// Token to complete, returns true if it returned before the timeout or
// returns false if the timeout occurred. In the case of a timeout the Token
// does not have an error set in case the caller wishes to wait again.
WaitTimeout(time.Duration) bool
// Done returns a channel that is closed when the flow associated
// with the Token completes. Clients should call Error after the
// channel is closed to check if the flow completed successfully.
//
// Done is provided for use in select statements. Simple use cases may
// use Wait or WaitTimeout.
Done() <-chan struct{}
Error() error
}
@ -52,21 +68,14 @@ type baseToken struct {
err error
}
// Wait will wait indefinitely for the Token to complete, ie the Publish
// to be sent and confirmed receipt from the broker
// Wait implements the Token Wait method.
func (b *baseToken) Wait() bool {
<-b.complete
return true
}
// WaitTimeout takes a time.Duration to wait for the flow associated with the
// Token to complete, returns true if it returned before the timeout or
// returns false if the timeout occurred. In the case of a timeout the Token
// does not have an error set in case the caller wishes to wait again
// WaitTimeout implements the Token WaitTimeout method.
func (b *baseToken) WaitTimeout(d time.Duration) bool {
b.m.Lock()
defer b.m.Unlock()
timer := time.NewTimer(d)
select {
case <-b.complete:
@ -80,6 +89,11 @@ func (b *baseToken) WaitTimeout(d time.Duration) bool {
return false
}
// Done implements the Token Done method.
func (b *baseToken) Done() <-chan struct{} {
return b.complete
}
func (b *baseToken) flowComplete() {
select {
case <-b.complete:
@ -125,7 +139,7 @@ type ConnectToken struct {
sessionPresent bool
}
// ReturnCode returns the acknowlegement code in the connack sent
// ReturnCode returns the acknowledgement code in the connack sent
// in response to a Connect()
func (c *ConnectToken) ReturnCode() byte {
c.m.RLock()
@ -160,6 +174,7 @@ type SubscribeToken struct {
baseToken
subs []string
subResult map[string]byte
messageID uint16
}
// Result returns a map of topics that were subscribed to along with
@ -175,6 +190,7 @@ func (s *SubscribeToken) Result() map[string]byte {
// required to provide information about calls to Unsubscribe()
type UnsubscribeToken struct {
baseToken
messageID uint16
}
// DisconnectToken is an extension of Token containing the extra fields

28
vendor/github.com/eclipse/paho.mqtt.golang/topic.go generated vendored
View File

@ -19,18 +19,18 @@ import (
"strings"
)
//ErrInvalidQos is the error returned when an packet is to be sent
//with an invalid Qos value
var ErrInvalidQos = errors.New("Invalid QoS")
// ErrInvalidQos is the error returned when an packet is to be sent
// with an invalid Qos value
var ErrInvalidQos = errors.New("invalid QoS")
//ErrInvalidTopicEmptyString is the error returned when a topic string
//is passed in that is 0 length
var ErrInvalidTopicEmptyString = errors.New("Invalid Topic; empty string")
// ErrInvalidTopicEmptyString is the error returned when a topic string
// is passed in that is 0 length
var ErrInvalidTopicEmptyString = errors.New("invalid Topic; empty string")
//ErrInvalidTopicMultilevel is the error returned when a topic string
//is passed in that has the multi level wildcard in any position but
//the last
var ErrInvalidTopicMultilevel = errors.New("Invalid Topic; multi-level wildcard must be last level")
// ErrInvalidTopicMultilevel is the error returned when a topic string
// is passed in that has the multi level wildcard in any position but
// the last
var ErrInvalidTopicMultilevel = errors.New("invalid Topic; multi-level wildcard must be last level")
// Topic Names and Topic Filters
// The MQTT v3.1.1 spec clarifies a number of ambiguities with regard
@ -46,10 +46,14 @@ var ErrInvalidTopicMultilevel = errors.New("Invalid Topic; multi-level wildcard
// - A TopicName may not contain a wildcard.
// - A TopicFilter may only have a # (multi-level) wildcard as the last level.
// - A TopicFilter may contain any number of + (single-level) wildcards.
// - A TopicFilter with a # will match the absense of a level
// - A TopicFilter with a # will match the absence of a level
// Example: a subscription to "foo/#" will match messages published to "foo".
func validateSubscribeMap(subs map[string]byte) ([]string, []byte, error) {
if len(subs) == 0 {
return nil, nil, errors.New("invalid subscription; subscribe map must not be empty")
}
var topics []string
var qoss []byte
for topic, qos := range subs {
@ -75,7 +79,7 @@ func validateTopicAndQos(topic string, qos byte) error {
}
}
if qos < 0 || qos > 2 {
if qos > 2 {
return ErrInvalidQos
}
return nil

2
vendor/github.com/eclipse/paho.mqtt.golang/trace.go generated vendored
View File

@ -27,7 +27,7 @@ type (
NOOPLogger struct{}
)
func (NOOPLogger) Println(v ...interface{}) {}
func (NOOPLogger) Println(v ...interface{}) {}
func (NOOPLogger) Printf(format string, v ...interface{}) {}
// Internal levels of library output that are initialised to not print

119
vendor/github.com/eclipse/paho.mqtt.golang/websocket.go generated vendored Normal file
View File

@ -0,0 +1,119 @@
package mqtt
import (
"crypto/tls"
"fmt"
"io"
"net"
"net/http"
"net/url"
"sync"
"time"
"github.com/gorilla/websocket"
)
// WebsocketOptions are config options for a websocket dialer
type WebsocketOptions struct {
ReadBufferSize int
WriteBufferSize int
Proxy ProxyFunction
}
type ProxyFunction func(req *http.Request) (*url.URL, error)
// NewWebsocket returns a new websocket and returns a net.Conn compatible interface using the gorilla/websocket package
func NewWebsocket(host string, tlsc *tls.Config, timeout time.Duration, requestHeader http.Header, options *WebsocketOptions) (net.Conn, error) {
if timeout == 0 {
timeout = 10 * time.Second
}
if options == nil {
// Apply default options
options = &WebsocketOptions{}
}
if options.Proxy == nil {
options.Proxy = http.ProxyFromEnvironment
}
dialer := &websocket.Dialer{
Proxy: options.Proxy,
HandshakeTimeout: timeout,
EnableCompression: false,
TLSClientConfig: tlsc,
Subprotocols: []string{"mqtt"},
ReadBufferSize: options.ReadBufferSize,
WriteBufferSize: options.WriteBufferSize,
}
ws, resp, err := dialer.Dial(host, requestHeader)
if err != nil {
if resp != nil {
WARN.Println(CLI, fmt.Sprintf("Websocket handshake failure. StatusCode: %d. Body: %s", resp.StatusCode, resp.Body))
}
return nil, err
}
wrapper := &websocketConnector{
Conn: ws,
}
return wrapper, err
}
// websocketConnector is a websocket wrapper so it satisfies the net.Conn interface so it is a
// drop in replacement of the golang.org/x/net/websocket package.
// Implementation guide taken from https://github.com/gorilla/websocket/issues/282
type websocketConnector struct {
*websocket.Conn
r io.Reader
rio sync.Mutex
wio sync.Mutex
}
// SetDeadline sets both the read and write deadlines
func (c *websocketConnector) SetDeadline(t time.Time) error {
if err := c.SetReadDeadline(t); err != nil {
return err
}
err := c.SetWriteDeadline(t)
return err
}
// Write writes data to the websocket
func (c *websocketConnector) Write(p []byte) (int, error) {
c.wio.Lock()
defer c.wio.Unlock()
err := c.WriteMessage(websocket.BinaryMessage, p)
if err != nil {
return 0, err
}
return len(p), nil
}
// Read reads the current websocket frame
func (c *websocketConnector) Read(p []byte) (int, error) {
c.rio.Lock()
defer c.rio.Unlock()
for {
if c.r == nil {
// Advance to next message.
var err error
_, c.r, err = c.NextReader()
if err != nil {
return 0, err
}
}
n, err := c.r.Read(p)
if err == io.EOF {
// At end of message.
c.r = nil
if n > 0 {
return n, nil
}
// No data read, continue to next message.
continue
}
return n, err
}
}