Restructure as monorepo and add flake packages/apps

- Move Elixir code to server/ subdirectory for monorepo structure
- Update flake.nix to provide packages and apps outputs for nix run support
- Update nix/package.nix to accept src parameter instead of fetchgit
- Add NixOS module export for easy consumption

Now supports: nix run, nix build, and nix develop from git repo

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
2025-08-02 19:54:17 -07:00
co-authored by Claude
parent 46e585ec92
commit b6769abbe9
197 changed files with 9555 additions and 7 deletions
Binary file not shown.
+425
View File
@@ -0,0 +1,425 @@
# Changelog
## 0.9.9 - 2021-06-14
- Switch to using `:ssl.handshake` instead of `:ssl.ssl_accept`
for OTP 24 compatibility.
## 0.9.8 - 2021-03-02
- Ease the requirement on the gen_state_machine dependency such
that is accept gen_state_machine 3.0.0
## 0.9.6 - 2020-12-01
- Accepted some open PRs that has been waiting for a long while. These
include allowing the user to pass in options to connections, and
will categorize a connection error as "connectivity", such that
tortoise will try to reconnect with back-off.
## 0.9.5 - 2020-10-20
### Fixed
- Allow a timeout on async publish; this will allow the user to
override the default timeout set when retrieving the connection;
Note that this timeout will not work for the publish itself.
## 0.9.4 - 2018-11-16
### Fixed
- A connection timeout is now categorized as a connectivity
issue. Thanks to [Riccardo Binetti](https://github.com/rbino) and
[Frank Hunleth](https://github.com/fhunleth) for this contribution.
## 0.9.3 - 2018-11-15
### Fixed
- Pass on connection errors so we can back-off on connection timeouts.
## 0.9.2 - 2018-09-01
### Added
- A use-macro has been added to `Tortoise.Handler` which implements
default behaviour for the callbacks so the user only have to
implement the callbacks they care about. This should improve
usability of the project quite a bit.
## 0.9.1 - 2018-08-31
### Fixed
- Avoid a possible race condition that could happen when a connection
was requested while the connection was in the process of being
established. This update should avoid hanging processes when a
publish is attempted in the small window between going from
connecting to connected.
### Added
- A usage example was added to the README-file. Thanks, [Tanweer
Shahzaad](https://github.com/tanweerdev).
## 0.9.0 - 2018-08-29
### Changed
- The `Tortoise` module does no longer delegate the `subscribe`,
`unsubscribe`, and their `_sync` variants to the
`Tortoise.Connection` module. They should just be accessed directly
on the `Tortoise.Connection` module instead. This is done to make
the documentation efforts easier, because conceptually the
subscription belongs to the connection.
### Fixed
- Improvements has been made to the type specs and the documentation
of various modules.
### Added
- A `ping/1` and `ping_sync/2` function has been added to the
`Tortoise.Connection` module. This allow the user to ping the broker
to check the connection. Notice that the connection process will
still send ping messages to fulfill the connection keep alive
timeout interval. There is no need to manually ping an open
connection.
## 0.8.3 - 2018-08-28
### Fixed
- Upgraded to a more recent version of dialyxir and made the project
pass the dialyzer.
## 0.8.2 - 2018-08-26
### Fixed
- As according to the specification, an unsubscribe, and a subscribe
package can no longer be encoded if it has no topic filters defined.
- Got rid of TravisCI in favor of SemaphoreCI. Hopefully this will
provide more reliable CI, and a funner experience for people working
on Tortoise.
## 0.8.1 - 2018-08-25
### Added
- `disconnect/1` has been added to the `Tortoise.Connection`
module. Given a `client_id` it will close down the connection
cleanly; in flight messages will get canceled and a disconnect
package will be send to the broker.
## 0.8.0 - 2018-08-19
### Changed
- When a connection drop the inflight manager will now re-transmit
packages with quality of service that was inflight when the regain a
connection.
- Improved the logic for handling incoming publishes with Quality of
Service 2 (exactly once delivery). The inflight manager will ensure
we only handle the message once (as we should); possible duplicate
messages caused by retransmisions should get trapped and not
forwarded to the user defined publish handler.
- It is now possible to pass in a binary/string as the host for a
connection using the `Tortoise.Transport.SSL` module; it will be
coerced to a charlist.
### Added
- The documentation on how to connect to a broker using the
`Tortoise.Transport.SSL` method has been improved by [Bram
Verburg](https://github.com/voltone).
## 0.7.0 - 2018-08-09
### Changed
- The controller will stop with an `{:protocol_violation,
{:unexpected_package_from_remote, package}}` if the server send a
package of the type `package`. This is the right thing to do because
servers should implement the protocol proper, and clients should not
accept unexpected packages.
- Instead of using the Logger module to log the ping response times
from the connection keep alive messages the ping response times are
not dispatched using a the ETS based PubSub. This will allow the
user to log, or do whatever they want with the ping response time.
- Expose connection status changes via the `Tortoise.Events` pubsub
making it possible to implement custom behavior when the connection
goes down and when it becomes available. Great for setting up
alarms.
- The receiver will now broadcast a "status down" message on the
`Tortoise.Events` pubsub, and the connection process will listen for
that message and enter the reconnect on status change.
### Removed
- `Tortoise.Connection.reconnect/1` has been removed as it is has been
replaced with the pubsub based status-down message approach
described in the "changed"-section. It might make a return at some
point, but for now it has been removed.
## 0.6.0 - 2018-07-29
### Changed
- Keep the active connection in a ETS based registry, allowing
processes that publish messages to the wire, using either
`Tortoise.publish/4` or `Tortoise.Pipe.publish/4` to obtain the
network connection and shoot the message directly on the network
socket. With this change we can also error out if a process tries to
send on a non-existing connection.
- The implementation for `Tortoise.publish/4` and
`Tortoise.publish_sync/4` has been moved to the `Tortoise` module
itself and is therefore no longer delegated to the
`Tortoise.Connection` module. This changes the interface a bit but
makes for a cleaner interface.
### Added
- This release introduces a `Tortoise.Events` module that implements a
`Registry` based PubSub. Tortoise will use this to dispatch system
events to listeners. For now it is used to dispatch new network
connections, which is currently used by `Tortoise.publish`, the
`Tortoise.Connection.Inflight` process, and the `Tortoise.Pipe`s. In
the future we might add more message types to `Tortoise.Events`.
- Tests for `Tortoise.publish/4` and `Tortoise.publish_sync/4`
### Removed
- The `Tortoise.Connection.Transmitter` process is no longer needed,
so it has been removed.
- Some dead code in the `Tortoise.Connection.Inflight` module has been
removed. This should not change anything user facing.
### Fixed
- A server sending a ping request to a client is now considered a
protocol violation, as specified by both the MQTT 3.1.1 and MQTT 5
specifications.
- The connection process will now cancel the keep alive timer if it
goes offline. Previously it would terminate itself because it would
not get the ping response from the server.
- Regression: The receiver will no longer crash on an assertion when
it request a reconnect from the `Tortoise.Connection` process.
- The specified Elixir version in the mix.exs file should now allow
more versions of Elixir without warnings.
## 0.5.1 - 2018-07-23
### Fixed
- The receiver could report an invalid header length protocol
violation when the server send too few bytes, and thus taking too
long sending the complete package header. This patch will ensure we
only report invalid header length when we have pulled more than 40
bits in.
## 0.5.0 - 2018-07-18
### Added
- Documentation. Lot's of documentation. It might be a bit wordy, and
it might repeat itself here and there, but it is a start that can be
improved upon and good enough for a 0.4.4 release.
- Added some articles that describe usage of Tortoise and MQTT.
### Changed
- rename `renew/1` to `reconnect/1` on `Tortoise.Connection`-module.
- Some type specs has been altered to better fit reality.
- The `Tortoise.Transport.Tcp` transport will now accept a binary as a
host name. Internally it will cast it to a char list.
## 0.4.3 - 2018-07-17
### Changed
- The wrong field was pulled out of the configuration options when the
last will message was set, so it was impossible to configure a last
will message. The last will message can now be set by passing in a
`%Tortoise.Package.Publish{}` struct as the `:will` when starting a
connection.
## 0.4.2 - 2018-07-08
### Changed
- The `Tortoise.Connection.renew/1` will now return `:ok` on success;
allowing the `Torotise.Connection.Receiver` to not crash on its
assertion when it request a reconnect.
## 0.4.1 - 2018-07-08
### Changed
- Tortoise should now survive the server it is connected to being
restarted. `{:error, :econnrefused}` and `{:error, :closed}` has
been added to the errors that make tortoise attempt a reconnect.
## 0.4.0 - 2018-07-08
### Added
- Incremental backoff has been added to the connector, allowing us to
retry reconnecting to the broker if the initial (or later reconnect)
attempt fails. The backoff will per default start retrying after 100
ms and it will increment in multiples of 2 up until 30 seconds, at
which point it will flip back to 100 ms and start over. This should
ensure that we will be able to connect fairly quickly if it is a
network fluke (or the network devise is not ready yet), and still
not try *too often* or *too quickly*.
The backoff can be configured by passing `backoff` containing a
keyword list to the connection specification. Example `backoff:
[min_interval: 100, max_interval: 30_000]`. Both times are in
milliseconds.
### Changed
- The code for establishing a connection and eventually reconnecting
has been combined into one. This makes it easier to test and verify,
and it will make it easier to handle connection errors.
Because the initial connection is happening outside of the `init/1`
function the possible return values of the
`Tortoise.Connection.start_link/1`-function has changed a bit. A
fatal error will cause the connection process to exit instead
because the init will always return `{:ok, state}`. This might break
some implementation using Tortoise.
For now it is only `{:error, :nxdomain}` that is handled with
connection retries. Error categorization has been planned so we can
error out if it is a non-recoverable error reason (such as no cacert
files specified), instead of retrying the connection. In the near
future more error reasons will be handled with reconnect attempts.
- A protocol violation from the server during connection will be
handled better; previously it would error with a decoding error,
because it would attempt to decode 4 random bytes. The error message
should be obvious now.
## 0.3.0 - 2018-06-10
### Added
- Thanks to [Troels Brødsgaard](https://github.com/trarbr) Tortoise
now implement a module for its registry. This is found in
`Tortoise.Registry`.
- The user defined controller handler callback module now accept "next
actions" in the return tuple; this allow the user to specify that a
topic should get subscribed to, or unsubscribed from, by specifying
a return like `{:ok, new_state, [{:subscribe, "foo/bar", qos: 3},
{:unsubscribe, "baz/quux"}]}`.
This is needed as the controller must not be blocked, and the user
defined callback module run in the context of the controller. By
allowing next actions like this the user can subscribe and
unsubscribe to topics when certain events happen.
- The test coverage tool will now ignore modules found in the
*lib/tortoise/handlers/*-folder. These modules implement the
`Tortoise.Handler`-behaviour, so they should be good.
### Changed
- `Tortoise.subscribe/3` is now async, so a message will get sent to
the mailbox of the calling process. The old behavior can be found in
the newly created `Tortoise.subscribe_sync/3` that will block until
the server has acknowledged the subscribe.
- `Tortoise.unsubscribe/3` is now also async, so like the subscribe a
message will get sent to the mailbox of the calling process. The old
behavior can be found in the newly added
`Tortoise.unsubscribe_sync/3` that will block until the server has
acknowledged the subscribe.
- A major refactorization of the code handling the logic running the
user defined controller callbacks has been lifted from the
`Tortoise.Connection.Controller` and put into the `Tortoise.Handler`
module. This change made it possible to support the next actions,
and makes it much easier test and add new next action types in the
future.
## 0.2.2 - 2018-05-29
### Changed
- Fix an issue where larger messages would crash the receiver. It has
been fixed and tested with messages as large as 268435455 bytes;
which is a pretty big MQTT message.
## 0.2.1 - 2018-05-29
### Added
- The `Tortoise.Transport.SSL` will now pass in `[verify:
:verify_peer]` as the default option when connecting. This will
guide the user to pass in a list of trusted CA certificates, such as
one provided by the Certifi package, or opt out of this by passing
the `verify: :verify_none` option; this will make it hard for the
user to make unsafe choices.
Thanks to [Bram Verburg](https://github.com/voltone) for this
improvement.
## 0.2.0 - 2018-05-28
### Added
- Experimental SSL support, please try it out and provide feedback.
- Abstract the network communication into a behaviour called
`Tortoise.Transport`. This behaviour specify callbacks needed to
connect, receive, and send messages using a network transport. It
also specify setting and getting options, as well as listening using
that network transport; the latter part is done so they can be used
in integration tests.
- A TCP transport has been created for communicating with a broker
using TCP. Use `Tortoise.Transport.Tcp` when specifying the server
in the connection to use the TCP transport.
- A SSL transport `Tortoise.Transport.SSL` has been added to the
project allowing us to connect to a broker using an encrypted
connection.
### Removed
- The `{:tcp, 'localhost', 1883}` connection specification has been
removed in favor of `{Tortoise.Transport.Tcp, host: 'localhost',
port: 1883}`. This is done because we support multiple transport
types now, such as the `Tortoise.Transport.SSL` type (which also
takes a `key` and a `cert` option). The format is `{transport,
opts}`.
## 0.1.0 - 2018-05-21
### Added
- The project is now on Hex which will hopefully broaden the user
base. Future changes will be logged to this file.
- We will from now on update the version number following Semantic
Versioning, and major changes should get logged to this file.
+201
View File
@@ -0,0 +1,201 @@
Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/
TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
1. Definitions.
"License" shall mean the terms and conditions for use, reproduction,
and distribution as defined by Sections 1 through 9 of this document.
"Licensor" shall mean the copyright owner or entity authorized by
the copyright owner that is granting the License.
"Legal Entity" shall mean the union of the acting entity and all
other entities that control, are controlled by, or are under common
control with that entity. For the purposes of this definition,
"control" means (i) the power, direct or indirect, to cause the
direction or management of such entity, whether by contract or
otherwise, or (ii) ownership of fifty percent (50%) or more of the
outstanding shares, or (iii) beneficial ownership of such entity.
"You" (or "Your") shall mean an individual or Legal Entity
exercising permissions granted by this License.
"Source" form shall mean the preferred form for making modifications,
including but not limited to software source code, documentation
source, and configuration files.
"Object" form shall mean any form resulting from mechanical
transformation or translation of a Source form, including but
not limited to compiled object code, generated documentation,
and conversions to other media types.
"Work" shall mean the work of authorship, whether in Source or
Object form, made available under the License, as indicated by a
copyright notice that is included in or attached to the work
(an example is provided in the Appendix below).
"Derivative Works" shall mean any work, whether in Source or Object
form, that is based on (or derived from) the Work and for which the
editorial revisions, annotations, elaborations, or other modifications
represent, as a whole, an original work of authorship. For the purposes
of this License, Derivative Works shall not include works that remain
separable from, or merely link (or bind by name) to the interfaces of,
the Work and Derivative Works thereof.
"Contribution" shall mean any work of authorship, including
the original version of the Work and any modifications or additions
to that Work or Derivative Works thereof, that is intentionally
submitted to Licensor for inclusion in the Work by the copyright owner
or by an individual or Legal Entity authorized to submit on behalf of
the copyright owner. For the purposes of this definition, "submitted"
means any form of electronic, verbal, or written communication sent
to the Licensor or its representatives, including but not limited to
communication on electronic mailing lists, source code control systems,
and issue tracking systems that are managed by, or on behalf of, the
Licensor for the purpose of discussing and improving the Work, but
excluding communication that is conspicuously marked or otherwise
designated in writing by the copyright owner as "Not a Contribution."
"Contributor" shall mean Licensor and any individual or Legal Entity
on behalf of whom a Contribution has been received by Licensor and
subsequently incorporated within the Work.
2. Grant of Copyright License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
copyright license to reproduce, prepare Derivative Works of,
publicly display, publicly perform, sublicense, and distribute the
Work and such Derivative Works in Source or Object form.
3. Grant of Patent License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
(except as stated in this section) patent license to make, have made,
use, offer to sell, sell, import, and otherwise transfer the Work,
where such license applies only to those patent claims licensable
by such Contributor that are necessarily infringed by their
Contribution(s) alone or by combination of their Contribution(s)
with the Work to which such Contribution(s) was submitted. If You
institute patent litigation against any entity (including a
cross-claim or counterclaim in a lawsuit) alleging that the Work
or a Contribution incorporated within the Work constitutes direct
or contributory patent infringement, then any patent licenses
granted to You under this License for that Work shall terminate
as of the date such litigation is filed.
4. Redistribution. You may reproduce and distribute copies of the
Work or Derivative Works thereof in any medium, with or without
modifications, and in Source or Object form, provided that You
meet the following conditions:
(a) You must give any other recipients of the Work or
Derivative Works a copy of this License; and
(b) You must cause any modified files to carry prominent notices
stating that You changed the files; and
(c) You must retain, in the Source form of any Derivative Works
that You distribute, all copyright, patent, trademark, and
attribution notices from the Source form of the Work,
excluding those notices that do not pertain to any part of
the Derivative Works; and
(d) If the Work includes a "NOTICE" text file as part of its
distribution, then any Derivative Works that You distribute must
include a readable copy of the attribution notices contained
within such NOTICE file, excluding those notices that do not
pertain to any part of the Derivative Works, in at least one
of the following places: within a NOTICE text file distributed
as part of the Derivative Works; within the Source form or
documentation, if provided along with the Derivative Works; or,
within a display generated by the Derivative Works, if and
wherever such third-party notices normally appear. The contents
of the NOTICE file are for informational purposes only and
do not modify the License. You may add Your own attribution
notices within Derivative Works that You distribute, alongside
or as an addendum to the NOTICE text from the Work, provided
that such additional attribution notices cannot be construed
as modifying the License.
You may add Your own copyright statement to Your modifications and
may provide additional or different license terms and conditions
for use, reproduction, or distribution of Your modifications, or
for any such Derivative Works as a whole, provided Your use,
reproduction, and distribution of the Work otherwise complies with
the conditions stated in this License.
5. Submission of Contributions. Unless You explicitly state otherwise,
any Contribution intentionally submitted for inclusion in the Work
by You to the Licensor shall be under the terms and conditions of
this License, without any additional terms or conditions.
Notwithstanding the above, nothing herein shall supersede or modify
the terms of any separate license agreement you may have executed
with Licensor regarding such Contributions.
6. Trademarks. This License does not grant permission to use the trade
names, trademarks, service marks, or product names of the Licensor,
except as required for reasonable and customary use in describing the
origin of the Work and reproducing the content of the NOTICE file.
7. Disclaimer of Warranty. Unless required by applicable law or
agreed to in writing, Licensor provides the Work (and each
Contributor provides its Contributions) 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. You are solely responsible for determining the
appropriateness of using or redistributing the Work and assume any
risks associated with Your exercise of permissions under this License.
8. Limitation of Liability. In no event and under no legal theory,
whether in tort (including negligence), contract, or otherwise,
unless required by applicable law (such as deliberate and grossly
negligent acts) or agreed to in writing, shall any Contributor be
liable to You for damages, including any direct, indirect, special,
incidental, or consequential damages of any character arising as a
result of this License or out of the use or inability to use the
Work (including but not limited to damages for loss of goodwill,
work stoppage, computer failure or malfunction, or any and all
other commercial damages or losses), even if such Contributor
has been advised of the possibility of such damages.
9. Accepting Warranty or Additional Liability. While redistributing
the Work or Derivative Works thereof, You may choose to offer,
and charge a fee for, acceptance of support, warranty, indemnity,
or other liability obligations and/or rights consistent with this
License. However, in accepting such obligations, You may act only
on Your own behalf and on Your sole responsibility, not on behalf
of any other Contributor, and only if You agree to indemnify,
defend, and hold each Contributor harmless for any liability
incurred by, or claims asserted against, such Contributor by reason
of your accepting any such warranty or additional liability.
END OF TERMS AND CONDITIONS
APPENDIX: How to apply the Apache License to your work.
To apply the Apache License to your work, attach the following
boilerplate notice, with the fields enclosed by brackets "{}"
replaced with your own identifying information. (Don't include
the brackets!) The text should be enclosed in the appropriate
comment syntax for the file format. We also recommend that a
file or class name and description of purpose be included on the
same "printed page" as the copyright notice for easier
identification within third-party archives.
Copyright 2018 Martin Gausby
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
+186
View File
@@ -0,0 +1,186 @@
# Tortoise
[![Build Status](https://semaphoreci.com/api/v1/gausby/tortoise/branches/master/badge.svg)](https://semaphoreci.com/gausby/tortoise)
[![Hex.pm](https://img.shields.io/hexpm/l/tortoise.svg "Apache 2.0 Licensed")](https://github.com/gausby/tortoise/blob/master/LICENSE)
[![Hex version](https://img.shields.io/hexpm/v/tortoise.svg "Hex version")](https://hex.pm/packages/tortoise)
[![Coverage Status](https://coveralls.io/repos/github/gausby/tortoise/badge.svg?branch=master)](https://coveralls.io/github/gausby/tortoise?branch=master)
A MQTT Client application that keep connections to one or more MQTT
brokers, handles subscriptions, and expose a publisher for publishing
messages to the broker.
Amongst other things Tortoise supports:
- Keeping a connection to a MQTT server (version 3.1.1 for now)
- Retry connecting with incremental back-off
- Publishing and subscribing to topics of QoS 0, 1, and 2
- Connections support last will message
- Connecting via TCP and SSL
- The fundamentals are there, but some of the API's might change in
the near future
- A PubSub system where one can listen to system events. For now
connection status and ping response times can be subscribed for
statistics and administrative purposes.
Most of the public facing interface should be in the `Tortoise`
module. See the GitHub issues for work in progress "known issues in
the design", "what needs to be done", and so forth; feel free to open
your own issues if something is confusing or broken.
I would love to get some feedback and help building this thing.
## Example
A supervised connection can be started like this:
``` elixir
# connect to the server and subscribe to foo/bar with QoS 0
Tortoise.Supervisor.start_child(
client_id: "my_client_id",
handler: {Tortoise.Handler.Logger, []},
server: {Tortoise.Transport.Tcp, host: 'localhost', port: 1883},
subscriptions: [{"foo/bar", 0}])
# publish a message on the broker
Tortoise.publish("my_client_id", "foo/bar", "Hello from the World of Tomorrow !", qos: 0)
```
To connect to a MQTT server using SSL the `Tortoise.Transport.SSL`
transport can be used. This requires configuration of the server's
CA certificate, and possibly a client certificate and key. For
example, when using the [certifi](https://hex.pm/packages/certifi)
package as the CA trust store:
``` elixir
Tortoise.Supervisor.start_child(
client_id: "smart-spoon",
handler: {Tortoise.Handler.Logger, []},
server: {
Tortoise.Transport.SSL,
host: host, port: port,
cacertfile: :certifi.cacertfile(),
key: key, cert: cert
},
subscriptions: [{"foo/bar", 0}])
```
Alternatively, for testing purposes, server certificate verification
can be disabled by passing `verify: :verify_none` in the server
options. In that case there is no need for CA certificates, but an
attacker could intercept the connection without detection!
Look at the `connection_test.exs`-file for more examples.
Example Handler
```elixir
defmodule Tortoise.Handler.Example do
use Tortoise.Handler
def init(args) do
{:ok, args}
end
def connection(status, state) do
# `status` will be either `:up` or `:down`; you can use this to
# inform the rest of your system if the connection is currently
# open or closed; tortoise should be busy reconnecting if you get
# a `:down`
{:ok, state}
end
# topic filter room/+/temp
def handle_message(["room", room, "temp"], payload, state) do
# :ok = Temperature.record(room, payload)
{:ok, state}
end
def handle_message(topic, payload, state) do
# unhandled message! You will crash if you subscribe to something
# and you don't have a 'catch all' matcher; crashing on unexpected
# messages could be a strategy though.
{:ok, state}
end
def subscription(status, topic_filter, state) do
{:ok, state}
end
def terminate(reason, state) do
# tortoise doesn't care about what you return from terminate/2,
# that is in alignment with other behaviours that implement a
# terminate-callback
:ok
end
end
```
## Upgrade path
### pre-0.9 to 0.9
The `subscribe/3`, `unsubscribe/3`, `subscribe_sync/3`, and
`unsubscribe_sync/3` is no longer exposed on the `Tortoise`
module. The functionality has been moved to the `Tortoise.Connection`
module. The functions has the same arities and functionality, so the
upgrade path is a simple search and replace:
- `"Tortoise.subscribe(" -> "Tortoise.Connection.subscribe("`
- `"Tortoise.subscribe_sync(" -> "Tortoise.Connection.subscribe_sync("`
- `"Tortoise.unsubscribe(" -> "Tortoise.Connection.unsubscribe("`
- `"Tortoise.unsubscribe_sync(" -> "Tortoise.Connection.unsubscribe_sync("`
This change is done because the `Tortoise.Connection` module should be
in charge of changes to the connection life-cycle.
## Installation
Tortoise can be installed by adding `tortoise` to your list of
dependencies in `mix.exs`:
```elixir
def deps do
[
{:tortoise, "~> 0.9"}
]
end
```
Documentation should be available at
[https://hexdocs.pm/tortoise](https://hexdocs.pm/tortoise).
## Development
To start developing, run the following commands:
```
mix deps.get
MIX_ENV=test mix eqc.install --mini
mix test
```
## Building documentation
To build the documentation run the following command in a terminal emulator:
``` shell
MIX_ENV=docs mix docs
```
This will build the documentation in place them in the *doc*-folder in
the root of the project. These docs will also find their way to the
Hexdocs website when the project is published on Hex.
## License
Copyright 2018 Martin Gausby
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
+43
View File
@@ -0,0 +1,43 @@
{<<"app">>,<<"tortoise">>}.
{<<"build_tools">>,[<<"mix">>]}.
{<<"description">>,<<"A MQTT client for Elixir.">>}.
{<<"elixir">>,<<"~> 1.6">>}.
{<<"files">>,
[<<"lib">>,<<"lib/tortoise">>,<<"lib/tortoise/handler">>,
<<"lib/tortoise/handler/default.ex">>,<<"lib/tortoise/handler/logger.ex">>,
<<"lib/tortoise/decodable.ex">>,<<"lib/tortoise/handler.ex">>,
<<"lib/tortoise/app.ex">>,<<"lib/tortoise/connection.ex">>,
<<"lib/tortoise/connection">>,<<"lib/tortoise/connection/receiver.ex">>,
<<"lib/tortoise/connection/controller.ex">>,
<<"lib/tortoise/connection/inflight.ex">>,
<<"lib/tortoise/connection/supervisor.ex">>,
<<"lib/tortoise/connection/inflight">>,
<<"lib/tortoise/connection/inflight/track.ex">>,
<<"lib/tortoise/connection/backoff.ex">>,<<"lib/tortoise/package.ex">>,
<<"lib/tortoise/pipe.ex">>,<<"lib/tortoise/package">>,
<<"lib/tortoise/package/publish.ex">>,
<<"lib/tortoise/package/pingresp.ex">>,
<<"lib/tortoise/package/unsubscribe.ex">>,
<<"lib/tortoise/package/pubrel.ex">>,<<"lib/tortoise/package/connect.ex">>,
<<"lib/tortoise/package/pingreq.ex">>,<<"lib/tortoise/package/meta.ex">>,
<<"lib/tortoise/package/suback.ex">>,<<"lib/tortoise/package/puback.ex">>,
<<"lib/tortoise/package/connack.ex">>,
<<"lib/tortoise/package/disconnect.ex">>,
<<"lib/tortoise/package/subscribe.ex">>,
<<"lib/tortoise/package/pubcomp.ex">>,<<"lib/tortoise/package/pubrec.ex">>,
<<"lib/tortoise/package/unsuback.ex">>,<<"lib/tortoise/supervisor.ex">>,
<<"lib/tortoise/transport">>,<<"lib/tortoise/transport/tcp.ex">>,
<<"lib/tortoise/transport/ssl.ex">>,<<"lib/tortoise/events.ex">>,
<<"lib/tortoise/encodable.ex">>,<<"lib/tortoise/transport.ex">>,
<<"lib/tortoise/registry.ex">>,<<"lib/tortoise.ex">>,<<"mix.exs">>,
<<"README.md">>,<<"CHANGELOG.md">>,<<"LICENSE">>]}.
{<<"licenses">>,[<<"Apache 2.0">>]}.
{<<"links">>,[{<<"GitHub">>,<<"https://github.com/gausby/tortoise">>}]}.
{<<"name">>,<<"tortoise">>}.
{<<"requirements">>,
[[{<<"app">>,<<"gen_state_machine">>},
{<<"name">>,<<"gen_state_machine">>},
{<<"optional">>,false},
{<<"repository">>,<<"hexpm">>},
{<<"requirement">>,<<"~> 2.0 or ~> 3.0">>}]]}.
{<<"version">>,<<"0.9.9">>}.
+343
View File
@@ -0,0 +1,343 @@
defmodule Tortoise do
@moduledoc """
A MQTT client for Elixir.
`Tortoise` provides ways of publishing messages to, and receiving
messages from one or many MQTT brokers via TCP or SSL. The design
philosophy of Tortoise is to hide the protocol specific details from
the user, and expose interfaces and a connection life cycle that
should feel natural to Elixir, while not limiting the capability of
what one can do with the MQTT protocol.
First off, connection to a broker happens through a connection
specification. This results in a process that can be supervised,
either by the application the connection should live and die with,
or by being supervised by the Tortoise application itself. Once the
connection is established the Tortoise application should do its
best to keep that connection open, by automatically sending keep
alive messages (as the protocol specifies), and eventually attempt
to reconnect if the connection should drop.
Secondly, a connection is specified with a user defined callback
module, following the `Tortoise.Handler`-behaviour, which allow the
user to hook into certain events happening in the life cycle of the
connection. This way code can get executed when:
- The connection is established
- The client has been disconnected from the broker
- A topic filter subscription has been accepted (or declined)
- A topic filter has been successfully unsubscribed
- A message is received on one of the subscribed topic filters
Besides this there are hooks for the usual life-cycle events one
would expect, such as `init/1` and `terminate/2`.
Thirdly, publishing is handled in such a way that the semantics of
the levels of Quality of Service, specified by the MQTT protocol, is
mapped to the Elixir message passing semantics. Tortoise expose an
interface for publishing messages that hide the protocol details of
message delivery (retrieval of acknowledge, release, complete
messages) and instead provide `Tortoise.publish/4` which will
deliver the message to the broker and receive a response in the
process mailbox when a message with a QoS>0 has been handed to the
server. This allow the user to keep track of the messages that has
been delivered, or simply by using the `Tortoise.publish_sync/4`
form that will block the calling process until the message has been
safely handed to the broker. Messages with QoS1 or QoS2 are stored
in a process until they are delivered, so once they are published
the client should retry delivery to make sure they reach their
destination.
An alternative way of posting messages is implemented in
`Tortoise.Pipe`, which provide a data structure that among other
things keep a reference to the connection socket. This allow for an
efficient way of posting messages because the data can get shot
directly onto the wire without having to copy the message between
processes (unless the message has a QoS of 1 or 2, in which case
they will end up in a process to ensure they will get
delivered). The pipe will automatically renew its connection socket
if the connection has been dropped, so ideally this message sending
approach should be fast and efficient.
"""
alias Tortoise.Package
alias Tortoise.Connection
alias Tortoise.Connection.Inflight
@typedoc """
An identifier used to identify the client on the server.
Most servers accept a maximum of 23 UTF-8 encode bytes for a client
id, and only the characters:
- "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
Tortoise accept atoms as client ids but they it will be converted to
a string before going on the wire. Be careful with atoms such as
`Example` because they are expanded to the atom `:"Elixir.Example"`,
it is really easy to hit the maximum byte limit. Solving this is
easy, just add a `:` before the client id such as `:Example`.
"""
@type client_id() :: atom() | String.t()
@typedoc """
A 16-bit number identifying a message in a message exchange.
Some MQTT packages are part of a message exchange and need an
identifier so the server and client can distinct between multiple
in-flight messages.
Tortoise will assign package identifier to packages that need them,
so outside of tests (where it is beneficial to assert on the
identifier of a package) it should be set by tortoise itself; so
just leave it as `nil`.
"""
@type package_identifier() :: 0x0001..0xFFFF | nil
@typedoc """
What Quality of Service (QoS) mode should be used.
Quality of Service is one of 0, 1, and 2 denoting the following:
- `0` no quality of service. The message is a fire and forget.
- `1` at least once delivery. The receiver will respond with an
acknowledge message, so the sender will be certain that the
message has reached the destination. It is possible that a message
will be delivered twice though, as the package identifier for a
publish will be relinquished when the message has been
acknowledged, so a package with the same identifier will be
treated as a new message though it might be a re-transmission.
- `2` exactly once delivery. The receiver will only receive the
message once. This happens by having a more elaborate message
exchange than the QoS=1 variant.
There are a difference in the semantics of assigning a QoS to a
publish and a subscription. When assigned to a publish the message
will get delivered to the server with the requested QoS; that is if
it accept that level of QoS for the given topic.
When used in the context of a subscription it should be read as *the
maximum QoS*. When messages are published to the subscribed topic
the message will get on-warded with the same topic as it was
delivered with, or downgraded to the maximum QoS of the subscription
for the given subscribing client. That is, if the client subscribe
with a maximum QoS=2 and a message is published to said topic with a
QoS=1, the message will get downgraded to QoS=1 when on-warded to
the client.
"""
@type qos() :: 0..2
@typedoc """
A topic for a message.
According to the MQTT 3.1.1 specification a valid topic must be at
least one character long. They are case sensitive and can include
space characters.
MQTT topics consist of topic levels which are delimited with forward
slashes `/`. A topic with a leading or trailing forward slash is
allowed but they create distinct topics from the ones without;
`/sports/tennis/results` are different from
`sports/tennis/results`. While a topic level normally require at
least one character the topic `/` (a single forward slash) is valid.
The server will drop the connection if it receive an invalid topic.
"""
@type topic() :: String.t()
@typedoc """
A topic filter for a subscription.
The topic filter is different from a `topic` because it is allowed
to contain wildcard characters:
- `+` is a single level wildcard which is allowed to stand on any
position in the topic filter. For instance: `sport/+/results` will
match `sport/tennis/results`, `sport/soccer/results`, etc.
- `#` is a multi-level wildcard and is only allowed to be on the
last position of the topic filter. For instance: `sport/#` will
match `sport/tennis/results`, `sport/tennis/announcements`, etc.
The server will reject any invalid topic filter and close the
connection.
"""
@type topic_filter() :: String.t()
@typedoc """
An optional message payload.
A message can optionally have a payload. The payload is a series of
bytes and for MQTT 3.1.1 the payload has no defined structure; any
series of bytes will do, and the client has to make sense of it.
The payload will be `nil` if there is no payload. This is done to
distinct between a zero byte binary and an empty payload.
"""
@type payload() :: binary() | nil
@doc """
Publish a message to the MQTT broker.
The publish function requires a `client_id` and a valid MQTT
topic. If no `payload` is set an empty zero byte message will get
send to the broker.
Optionally an options list can get passed to the publish, making it
possible to specify if the message should be retained on the server,
and with what quality of service the message should be published
with.
* `retain` indicates, when set to `true`, that the broker should
retain the message for the topic. Retained messages are
delivered to clients when they subscribe to the topic. Only one
message at a time can be retained for a given topic, so sending
a new one will overwrite the old. `retain` defaults to `false`.
* `qos` set the quality of service, and integer of 0, 1, or 2. The
`qos` defaults to `0`.
Publishing a message with the payload *hello* to to topic *foo/bar*
with a *QoS1* could look like this:
Tortoise.publish("client_id", "foo/bar", "hello", qos: 1)
Notice that if you want to send a message with an empty payload with
options you will have to set to payload to nil like this:
Tortoise.publish("client_id", "foo/bar", nil, retain: true)
## Return Values
The specified Quality of Service for a given publish will alter the
behaviour of the return value. When publishing a message with a QoS0
an `:ok` will simply get returned. This is because a QoS0 is a "fire
and forget." There are no quality of service so no efforts are made
to ensure that the message will reach its destination (though it very
likely will).
:ok = Tortoise.publish("client_id", "foo/bar", nil, qos: 0)
When a message is published using either a QoS1 or QoS2, Tortoise
will ensure that the message is delivered. A unique reference will
get returned and eventually a message will get delivered to the
process mailbox, containing the result of the publish when it has
been handed over:
{:ok, ref} = Tortoise.publish("client_id", "foo/bar", nil, qos: 2)
receive do
{{Tortoise, "client_id"}, ^ref, result} ->
IO.inspect({:result, result})
after
5000 ->
{:error, :timeout}
end
Be sure to implement a `handle_info/2` in `GenServer` processes that
publish messages using Tortoise.publish/4. Notice that the returned
message has a structure:
{{Tortoise, "client_id"}, ^ref, result}
It is possible to send to multiple clients and blanket match on
results designated for a given client id, and the message is tagged
with `Tortoise` so it is easy to see where the message originated
from.
"""
@spec publish(client_id(), topic(), payload, [options]) ::
:ok | {:ok, reference()} | {:error, :unknown_connection}
when payload: binary() | nil,
options:
{:qos, qos()}
| {:retain, boolean()}
| {:identifier, package_identifier()}
| {:timeout, non_neg_integer()}
def publish(client_id, topic, payload \\ nil, opts \\ []) do
qos = Keyword.get(opts, :qos, 0)
publish = %Package.Publish{
topic: topic,
qos: qos,
payload: payload,
retain: Keyword.get(opts, :retain, false)
}
timeout = Keyword.get(opts, :timeout, :infinity)
with {:ok, {transport, socket}} <- Connection.connection(client_id, timeout: timeout) do
case publish do
%Package.Publish{qos: 0} ->
encoded_publish = Package.encode(publish)
apply(transport, :send, [socket, encoded_publish])
%Package.Publish{qos: qos} when qos in [1, 2] ->
Inflight.track(client_id, {:outgoing, publish})
end
else
{:error, :unknown_connection} ->
{:error, :unknown_connection}
end
end
@doc """
Synchronously send a message to the MQTT broker.
This is very similar to `Tortoise.publish/4` with the difference
that it will block the calling process until the message has been
handed over to the server; the configuration options are the same
with the addition of the `timeout` option which specifies how long
we are willing to wait for a reply. Per default the timeout is set
to `:infinity`, it is advisable to set it to a reasonable amount in
milliseconds as it otherwise could block forever.
msg = "Hello, from the World of Tomorrow !"
case Tortoise.publish_sync("my_client_id", "foo/bar", msg, qos: 2, timeout: 200) do
:ok ->
:done
{:error, :timeout} ->
:timeout
end
Notice: It does not make sense to use `publish_sync/4` on a publish
that has a QoS=0, because that will return instantly anyways. It is
made possible for consistency, and it is the default QoS.
See the documentation for `Tortoise.publish/4` for configuration.
"""
@spec publish_sync(client_id(), topic(), payload, [options]) ::
:ok | {:error, :unknown_connection}
when payload: binary() | nil,
options:
{:qos, qos()}
| {:retain, boolean()}
| {:identifier, package_identifier()}
| {:timeout, timeout()}
def publish_sync(client_id, topic, payload \\ nil, opts \\ []) do
timeout = Keyword.get(opts, :timeout, :infinity)
qos = Keyword.get(opts, :qos, 0)
publish = %Package.Publish{
topic: topic,
qos: qos,
payload: payload,
retain: Keyword.get(opts, :retain, false)
}
with {:ok, {transport, socket}} <- Connection.connection(client_id, timeout: timeout) do
case publish do
%Package.Publish{qos: 0} ->
encoded_publish = Package.encode(publish)
apply(transport, :send, [socket, encoded_publish])
%Package.Publish{qos: qos} when qos in [1, 2] ->
Inflight.track_sync(client_id, {:outgoing, publish}, timeout)
end
else
{:error, :unknown_connection} ->
{:error, :unknown_connection}
end
end
end
+20
View File
@@ -0,0 +1,20 @@
defmodule Tortoise.App do
@moduledoc false
use Application
@impl true
def start(_type, _args) do
# read configuration and start connections
# start with client_id, and handler from config
children = [
{Registry, [keys: :unique, name: Tortoise.Registry]},
{Registry, [keys: :duplicate, name: Tortoise.Events]},
{Tortoise.Supervisor, [strategy: :one_for_one]}
]
opts = [strategy: :one_for_one, name: Tortoise]
Supervisor.start_link(children, opts)
end
end
@@ -0,0 +1,666 @@
defmodule Tortoise.Connection do
@moduledoc """
Establish a connection to a MQTT broker.
Todo.
"""
use GenServer
require Logger
defstruct [:client_id, :connect, :server, :status, :backoff, :subscriptions, :keep_alive, :opts]
alias __MODULE__, as: State
alias Tortoise.{Transport, Connection, Package, Events}
alias Tortoise.Connection.{Inflight, Controller, Receiver, Backoff}
alias Tortoise.Package.{Connect, Connack}
@doc """
Start a connection process and link it to the current process.
Read the documentation on `child_spec/1` if you want... (todo!)
"""
@spec start_link(options, GenServer.options()) :: GenServer.on_start()
when option:
{:client_id, Tortoise.client_id()}
| {:server, {atom(), term()}}
| {:user_name, String.t()}
| {:password, String.t()}
| {:keep_alive, non_neg_integer()}
| {:will, Tortoise.Package.Publish.t()}
| {:subscriptions,
[{Tortoise.topic_filter(), Tortoise.qos()}] | Tortoise.Package.Subscribe.t()}
| {:handler, {atom(), term()}},
options: [option]
def start_link(connection_opts, opts \\ []) do
client_id = Keyword.fetch!(connection_opts, :client_id)
server = connection_opts |> Keyword.fetch!(:server) |> Transport.new()
connect = %Package.Connect{
client_id: client_id,
user_name: Keyword.get(connection_opts, :user_name),
password: Keyword.get(connection_opts, :password),
keep_alive: Keyword.get(connection_opts, :keep_alive, 60),
will: Keyword.get(connection_opts, :will),
# if we re-spawn from here it means our state is gone
clean_session: true
}
backoff = Keyword.get(connection_opts, :backoff, [])
# This allow us to either pass in a list of topics, or a
# subscription struct. Passing in a subscription struct is helpful
# in tests.
subscriptions =
case Keyword.get(connection_opts, :subscriptions, []) do
topics when is_list(topics) ->
Enum.into(topics, %Package.Subscribe{})
%Package.Subscribe{} = subscribe ->
subscribe
end
# @todo, validate that the handler is valid
connection_opts = Keyword.take(connection_opts, [:client_id, :handler])
initial = {server, connect, backoff, subscriptions, connection_opts}
opts = Keyword.merge(opts, name: via_name(client_id))
GenServer.start_link(__MODULE__, initial, opts)
end
@doc false
@spec via_name(Tortoise.client_id()) ::
pid() | {:via, Registry, {Tortoise.Registry, {atom(), Tortoise.client_id()}}}
def via_name(client_id) do
Tortoise.Registry.via_name(__MODULE__, client_id)
end
@spec child_spec(Keyword.t()) :: %{
id: term(),
start: {__MODULE__, :start_link, [Keyword.t()]},
restart: :transient | :permanent | :temporary,
type: :worker
}
def child_spec(opts) do
%{
id: Keyword.get(opts, :name, __MODULE__),
start: {__MODULE__, :start_link, [opts]},
restart: Keyword.get(opts, :restart, :transient),
type: :worker
}
end
@doc """
Close the connection to the broker.
Given the `client_id` of a running connection it will cancel the
inflight messages and send the proper disconnect message to the
broker. The session will get terminated on the server.
"""
@spec disconnect(Tortoise.client_id()) :: :ok
def disconnect(client_id) do
GenServer.call(via_name(client_id), :disconnect)
end
@doc """
Return the list of subscribed topics.
Given the `client_id` of a running connection return its current
subscriptions. This is helpful in a debugging situation.
"""
@spec subscriptions(Tortoise.client_id()) :: Tortoise.Package.Subscribe.t()
def subscriptions(client_id) do
GenServer.call(via_name(client_id), :subscriptions)
end
@doc """
Subscribe to one or more topics using topic filters on `client_id`
The topic filter should be a 2-tuple, `{topic_filter, qos}`, where
the `topic_filter` is a valid MQTT topic filter, and `qos` an
integer value 0 through 2.
Multiple topics can be given as a list.
The subscribe function is asynchronous, so it will return `{:ok,
ref}`. Eventually a response will get delivered to the process
mailbox, tagged with the reference stored in `ref`. It will take the
form of:
{{Tortoise, ^client_id}, ^ref, ^result}
Where the `result` can be one of `:ok`, or `{:error, reason}`.
Read the documentation for `Tortoise.Connection.subscribe_sync/3`
for a blocking version of this call.
"""
@spec subscribe(Tortoise.client_id(), topic | topics, [options]) :: {:ok, reference()}
when topics: [topic],
topic: {Tortoise.topic_filter(), Tortoise.qos()},
options:
{:timeout, timeout()}
| {:identifier, Tortoise.package_identifier()}
def subscribe(client_id, topics, opts \\ [])
def subscribe(client_id, [{_, n} | _] = topics, opts) when is_number(n) do
caller = {_, ref} = {self(), make_ref()}
{identifier, opts} = Keyword.pop_first(opts, :identifier, nil)
subscribe = Enum.into(topics, %Package.Subscribe{identifier: identifier})
GenServer.cast(via_name(client_id), {:subscribe, caller, subscribe, opts})
{:ok, ref}
end
def subscribe(client_id, {_, n} = topic, opts) when is_number(n) do
subscribe(client_id, [topic], opts)
end
def subscribe(client_id, topic, opts) when is_binary(topic) do
case Keyword.pop_first(opts, :qos) do
{nil, _opts} ->
throw("Please specify a quality of service for the subscription")
{qos, opts} when qos in 0..2 ->
subscribe(client_id, [{topic, qos}], opts)
end
end
@doc """
Subscribe to topics and block until the server acknowledges.
This is a synchronous version of the
`Tortoise.Connection.subscribe/3`. In fact it calls into
`Tortoise.Connection.subscribe/3` but will handle the selective
receive loop, making it much easier to work with. Also, this
function can be used to block a process that cannot continue before
it has a subscription to the given topics.
See `Tortoise.Connection.subscribe/3` for configuration options.
"""
@spec subscribe_sync(Tortoise.client_id(), topic | topics, [options]) ::
:ok | {:error, :timeout}
when topics: [topic],
topic: {Tortoise.topic_filter(), Tortoise.qos()},
options:
{:timeout, timeout()}
| {:identifier, Tortoise.package_identifier()}
def subscribe_sync(client_id, topics, opts \\ [])
def subscribe_sync(client_id, [{_, n} | _] = topics, opts) when is_number(n) do
timeout = Keyword.get(opts, :timeout, 5000)
{:ok, ref} = subscribe(client_id, topics, opts)
receive do
{{Tortoise, ^client_id}, ^ref, result} -> result
after
timeout ->
{:error, :timeout}
end
end
def subscribe_sync(client_id, {_, n} = topic, opts) when is_number(n) do
subscribe_sync(client_id, [topic], opts)
end
def subscribe_sync(client_id, topic, opts) when is_binary(topic) do
case Keyword.pop_first(opts, :qos) do
{nil, _opts} ->
throw("Please specify a quality of service for the subscription")
{qos, opts} ->
subscribe_sync(client_id, [{topic, qos}], opts)
end
end
@doc """
Unsubscribe from one of more topic filters. The topic filters are
given as strings. Multiple topic filters can be given at once by
passing in a list of strings.
Tortoise.Connection.unsubscribe(client_id, ["foo/bar", "quux"])
This operation is asynchronous. When the operation is done a message
will be received in mailbox of the originating process.
"""
@spec unsubscribe(Tortoise.client_id(), topic | topics, [options]) :: {:ok, reference()}
when topics: [topic],
topic: Tortoise.topic_filter(),
options:
{:timeout, timeout()}
| {:identifier, Tortoise.package_identifier()}
def unsubscribe(client_id, topics, opts \\ [])
def unsubscribe(client_id, [topic | _] = topics, opts) when is_binary(topic) do
caller = {_, ref} = {self(), make_ref()}
{identifier, opts} = Keyword.pop_first(opts, :identifier, nil)
unsubscribe = %Package.Unsubscribe{identifier: identifier, topics: topics}
GenServer.cast(via_name(client_id), {:unsubscribe, caller, unsubscribe, opts})
{:ok, ref}
end
def unsubscribe(client_id, topic, opts) when is_binary(topic) do
unsubscribe(client_id, [topic], opts)
end
@doc """
Unsubscribe from topics and block until the server acknowledges.
This is a synchronous version of
`Tortoise.Connection.unsubscribe/3`. It will block until the server
has send the acknowledge message.
See `Tortoise.Connection.unsubscribe/3` for configuration options.
"""
@spec unsubscribe_sync(Tortoise.client_id(), topic | topics, [options]) ::
:ok | {:error, :timeout}
when topics: [topic],
topic: Tortoise.topic_filter(),
options:
{:timeout, timeout()}
| {:identifier, Tortoise.package_identifier()}
def unsubscribe_sync(client_id, topics, opts \\ [])
def unsubscribe_sync(client_id, topics, opts) when is_list(topics) do
timeout = Keyword.get(opts, :timeout, 5000)
{:ok, ref} = unsubscribe(client_id, topics, opts)
receive do
{{Tortoise, ^client_id}, ^ref, result} -> result
after
timeout ->
{:error, :timeout}
end
end
def unsubscribe_sync(client_id, topic, opts) when is_binary(topic) do
unsubscribe_sync(client_id, [topic], opts)
end
@doc """
Ping the broker.
When the round-trip is complete a message with the time taken in
milliseconds will be send to the process that invoked the ping
command.
The connection will automatically ping the broker at the interval
specified in the connection configuration, so there is no need to
setup a reoccurring ping. This ping function is exposed for
debugging purposes. If ping latency over time is desired it is
better to listen on `:ping_response` using the `Tortoise.Events`
PubSub.
"""
@spec ping(Tortoise.client_id()) :: {:ok, reference()}
defdelegate ping(client_id), to: Tortoise.Connection.Controller
@doc """
Ping the server and await the ping latency reply.
Takes a `client_id` and an optional `timeout`.
Like `ping/1` but will block the caller process until a response is
received from the server. The response will contain the ping latency
in milliseconds. The `timeout` defaults to `:infinity`, so it is
advisable to specify a reasonable time one is willing to wait for a
response.
"""
@spec ping_sync(Tortoise.client_id(), timeout()) :: {:ok, reference()} | {:error, :timeout}
defdelegate ping_sync(client_id, timeout \\ :infinity),
to: Tortoise.Connection.Controller
@doc false
@spec connection(Tortoise.client_id(), [opts]) ::
{:ok, {module(), term()}} | {:error, :unknown_connection} | {:error, :timeout}
when opts: {:timeout, timeout()} | {:active, boolean()}
def connection(client_id, opts \\ [active: false]) do
# register a connection subscription in the case we are currently
# in the connect phase; this solves a possible race condition
# where the connection is requested while the status is
# connecting, but will reach the receive block after the message
# has been dispatched from the pubsub; previously we registered
# for the connection message in this window.
{:ok, _} = Events.register(client_id, :connection)
case Tortoise.Registry.meta(via_name(client_id)) do
{:ok, {_transport, _socket} = connection} ->
{:ok, connection}
{:ok, :connecting} ->
timeout = Keyword.get(opts, :timeout, :infinity)
receive do
{{Tortoise, ^client_id}, :connection, {transport, socket}} ->
{:ok, {transport, socket}}
after
timeout ->
{:error, :timeout}
end
:error ->
{:error, :unknown_connection}
end
after
# if the connection subscription is non-active we should remove it
# from the registry, so the process will not receive connection
# messages when the connection is reestablished.
active? = Keyword.get(opts, :active, false)
unless active?, do: Events.unregister(client_id, :connection)
end
# Callbacks
@impl true
def init(
{transport, %Connect{client_id: client_id} = connect, backoff_opts, subscriptions, opts}
) do
state = %State{
client_id: client_id,
server: transport,
connect: connect,
backoff: Backoff.new(backoff_opts),
subscriptions: subscriptions,
opts: opts,
status: :down
}
Tortoise.Registry.put_meta(via_name(client_id), :connecting)
Tortoise.Events.register(client_id, :status)
# eventually, switch to handle_continue
send(self(), :connect)
{:ok, state}
end
@impl true
def terminate(_reason, state) do
:ok = Tortoise.Registry.delete_meta(via_name(state.connect.client_id))
:ok = Events.dispatch(state.client_id, :status, :terminated)
:ok
end
@impl true
def handle_info(:connect, state) do
# make sure we will not fall for a keep alive timeout while we reconnect
state = cancel_keep_alive(state)
with {%Connack{status: :accepted} = connack, socket} <-
do_connect(state.server, state.connect),
{:ok, state} = init_connection(socket, state) do
# we are connected; reset backoff state, etc
state =
%State{state | backoff: Backoff.reset(state.backoff)}
|> update_connection_status(:up)
|> reset_keep_alive()
case connack do
%Connack{session_present: true} ->
{:noreply, state}
%Connack{session_present: false} ->
:ok = Inflight.reset(state.client_id)
unless Enum.empty?(state.subscriptions), do: send(self(), :subscribe)
{:noreply, state}
end
else
%Connack{status: {:refused, reason}} ->
{:stop, {:connection_failed, reason}, state}
{:error, reason} ->
{timeout, state} = Map.get_and_update(state, :backoff, &Backoff.next/1)
case categorize_error(reason) do
:connectivity ->
Process.send_after(self(), :connect, timeout)
{:noreply, state}
:other ->
{:stop, reason, state}
end
end
end
def handle_info(:subscribe, %State{subscriptions: subscriptions} = state) do
client_id = state.connect.client_id
case Enum.empty?(subscriptions) do
true ->
# nothing to subscribe to, just continue
{:noreply, state}
false ->
# subscribe to the predefined topics
case Inflight.track_sync(client_id, {:outgoing, subscriptions}, 5000) do
{:error, :timeout} ->
{:stop, :subscription_timeout, state}
result ->
case handle_suback_result(result, state) do
{:ok, updated_state} ->
{:noreply, updated_state}
{:error, reasons} ->
error = {:unable_to_subscribe, reasons}
{:stop, error, state}
end
end
end
end
def handle_info(:ping, %State{} = state) do
case Controller.ping_sync(state.connect.client_id, 5000) do
{:ok, round_trip_time} ->
Events.dispatch(state.connect.client_id, :ping_response, round_trip_time)
state = reset_keep_alive(state)
{:noreply, state}
{:error, :timeout} ->
{:stop, :ping_timeout, state}
end
end
# dropping connection
def handle_info({transport, _socket}, state) when transport in [:tcp_closed, :ssl_closed] do
Logger.error("Socket closed before we handed it to the receiver")
# communicate that we are down
:ok = Events.dispatch(state.client_id, :status, :down)
{:noreply, state}
end
# react to connection status change events
def handle_info(
{{Tortoise, client_id}, :status, status},
%{client_id: client_id, status: current} = state
) do
case status do
^current ->
{:noreply, state}
:up ->
{:noreply, %State{state | status: status}}
:down ->
send(self(), :connect)
{:noreply, %State{state | status: status}}
end
end
@impl true
def handle_call(:subscriptions, _from, state) do
{:reply, state.subscriptions, state}
end
def handle_call(:disconnect, from, state) do
:ok = Events.dispatch(state.client_id, :status, :terminating)
:ok = Inflight.drain(state.client_id)
:ok = Controller.stop(state.client_id)
:ok = GenServer.reply(from, :ok)
{:stop, :shutdown, state}
end
@impl true
def handle_cast({:subscribe, {caller_pid, ref}, subscribe, opts}, state) do
client_id = state.connect.client_id
timeout = Keyword.get(opts, :timeout, 5000)
case Inflight.track_sync(client_id, {:outgoing, subscribe}, timeout) do
{:error, :timeout} = error ->
send(caller_pid, {{Tortoise, client_id}, ref, error})
{:noreply, state}
result ->
case handle_suback_result(result, state) do
{:ok, updated_state} ->
send(caller_pid, {{Tortoise, client_id}, ref, :ok})
{:noreply, updated_state}
{:error, reasons} ->
error = {:unable_to_subscribe, reasons}
send(caller_pid, {{Tortoise, client_id}, ref, {:error, reasons}})
{:stop, error, state}
end
end
end
def handle_cast({:unsubscribe, {caller_pid, ref}, unsubscribe, opts}, state) do
client_id = state.connect.client_id
timeout = Keyword.get(opts, :timeout, 5000)
case Inflight.track_sync(client_id, {:outgoing, unsubscribe}, timeout) do
{:error, :timeout} = error ->
send(caller_pid, {{Tortoise, client_id}, ref, error})
{:noreply, state}
unsubbed ->
topics = Keyword.drop(state.subscriptions.topics, unsubbed)
subscriptions = %Package.Subscribe{state.subscriptions | topics: topics}
send(caller_pid, {{Tortoise, client_id}, ref, :ok})
{:noreply, %State{state | subscriptions: subscriptions}}
end
end
# Helpers
defp handle_suback_result(%{:error => []} = results, %State{} = state) do
subscriptions = Enum.into(results[:ok], state.subscriptions)
{:ok, %State{state | subscriptions: subscriptions}}
end
defp handle_suback_result(%{:error => errors}, %State{}) do
{:error, errors}
end
defp reset_keep_alive(%State{keep_alive: nil} = state) do
ref = Process.send_after(self(), :ping, state.connect.keep_alive * 1000)
%State{state | keep_alive: ref}
end
defp reset_keep_alive(%State{keep_alive: previous_ref} = state) do
# Cancel the previous timer, just in case one was already set
_ = Process.cancel_timer(previous_ref)
ref = Process.send_after(self(), :ping, state.connect.keep_alive * 1000)
%State{state | keep_alive: ref}
end
defp cancel_keep_alive(%State{keep_alive: nil} = state) do
state
end
defp cancel_keep_alive(%State{keep_alive: keep_alive_ref} = state) do
_ = Process.cancel_timer(keep_alive_ref)
%State{state | keep_alive: nil}
end
# dispatch connection status if the connection status change
defp update_connection_status(%State{status: same} = state, same) do
state
end
defp update_connection_status(%State{} = state, status) do
:ok = Events.dispatch(state.connect.client_id, :status, status)
%State{state | status: status}
end
defp do_connect(server, %Connect{} = connect) do
%Transport{type: transport, host: host, port: port, opts: opts} = server
with {:ok, socket} <- transport.connect(host, port, opts, 10000),
:ok = transport.send(socket, Package.encode(connect)),
{:ok, packet} <- transport.recv(socket, 4, 5000) do
try do
case Package.decode(packet) do
%Connack{status: :accepted} = connack ->
{connack, socket}
%Connack{status: {:refused, _reason}} = connack ->
connack
end
catch
:error, {:badmatch, _unexpected} ->
violation = %{expected: Connect, got: packet}
{:error, {:protocol_violation, violation}}
end
else
{:error, :econnrefused} ->
{:error, {:connection_refused, host, port}}
{:error, :nxdomain} ->
{:error, {:nxdomain, host, port}}
{:error, {:options, {:cacertfile, []}}} ->
{:error, :no_cacertfile_specified}
{:error, :closed} ->
{:error, :server_closed_connection}
{:error, :timeout} ->
{:error, :connection_timeout}
{:error, other} ->
{:error, other}
end
end
defp init_connection(socket, %State{opts: opts, server: transport, connect: connect} = state) do
connection = {transport.type, socket}
:ok = start_connection_supervisor(opts)
:ok = Receiver.handle_socket(connect.client_id, connection)
:ok = Tortoise.Registry.put_meta(via_name(connect.client_id), connection)
:ok = Events.dispatch(connect.client_id, :connection, connection)
# set clean session to false for future reconnect attempts
connect = %Connect{connect | clean_session: false}
{:ok, %State{state | connect: connect}}
end
defp start_connection_supervisor(opts) do
case Connection.Supervisor.start_link(opts) do
{:ok, _pid} ->
:ok
{:error, {:already_started, _pid}} ->
:ok
end
end
defp categorize_error({:nxdomain, _host, _port}) do
:connectivity
end
defp categorize_error({:connection_refused, _host, _port}) do
:connectivity
end
defp categorize_error(:server_closed_connection) do
:connectivity
end
defp categorize_error(:connection_timeout) do
:connectivity
end
defp categorize_error(:enetunreach) do
:connectivity
end
defp categorize_error(_other) do
:other
end
end
@@ -0,0 +1,36 @@
defmodule Tortoise.Connection.Backoff do
@moduledoc false
defstruct min_interval: 100, max_interval: 30_000, value: nil
alias __MODULE__, as: State
@doc """
Create an opaque data structure that describe a incremental
back-off.
"""
def new(opts) do
min_interval = Keyword.get(opts, :min_interval, 100)
max_interval = Keyword.get(opts, :max_interval, 30_000)
%State{min_interval: min_interval, max_interval: max_interval}
end
def next(%State{value: nil} = state) do
current = state.min_interval
{current, %State{state | value: current}}
end
def next(%State{max_interval: same, value: same} = state) do
current = state.min_interval
{current, %State{state | value: current}}
end
def next(%State{value: value} = state) do
current = min(value * 2, state.max_interval)
{current, %State{state | value: current}}
end
def reset(%State{} = state) do
%State{state | value: nil}
end
end
@@ -0,0 +1,355 @@
defmodule Tortoise.Connection.Controller do
@moduledoc false
require Logger
alias Tortoise.{Package, Connection, Handler}
alias Tortoise.Connection.Inflight
alias Tortoise.Package.{
Connect,
Connack,
Disconnect,
Publish,
Puback,
Pubrec,
Pubrel,
Pubcomp,
Subscribe,
Suback,
Unsubscribe,
Unsuback,
Pingreq,
Pingresp
}
use GenServer
@enforce_keys [:client_id, :handler]
defstruct client_id: nil,
ping: :queue.new(),
status: :down,
awaiting: %{},
handler: %Handler{module: Handler.Default, initial_args: []}
alias __MODULE__, as: State
# Client API
def start_link(opts) do
client_id = Keyword.fetch!(opts, :client_id)
handler = Handler.new(Keyword.fetch!(opts, :handler))
init_state = %State{
client_id: client_id,
handler: handler
}
GenServer.start_link(__MODULE__, init_state, name: via_name(client_id))
end
defp via_name(client_id) do
Tortoise.Registry.via_name(__MODULE__, client_id)
end
def stop(client_id) do
GenServer.stop(via_name(client_id))
end
def info(client_id) do
GenServer.call(via_name(client_id), :info)
end
@spec ping(Tortoise.client_id()) :: {:ok, reference()}
def ping(client_id) do
ref = make_ref()
:ok = GenServer.cast(via_name(client_id), {:ping, {self(), ref}})
{:ok, ref}
end
@spec ping_sync(Tortoise.client_id(), timeout()) :: {:ok, reference()} | {:error, :timeout}
def ping_sync(client_id, timeout \\ :infinity) do
{:ok, ref} = ping(client_id)
receive do
{Tortoise, {:ping_response, ^ref, round_trip_time}} ->
{:ok, round_trip_time}
after
timeout ->
{:error, :timeout}
end
end
@doc false
def handle_incoming(client_id, package) do
GenServer.cast(via_name(client_id), {:incoming, package})
end
@doc false
def handle_result(client_id, {{pid, ref}, Package.Publish, result}) do
send(pid, {{Tortoise, client_id}, ref, result})
:ok
end
def handle_result(client_id, {{pid, ref}, type, result}) do
send(pid, {{Tortoise, client_id}, ref, result})
GenServer.cast(via_name(client_id), {:result, {type, result}})
end
@doc false
def handle_onward(client_id, %Package.Publish{} = publish) do
GenServer.cast(via_name(client_id), {:onward, publish})
end
# Server callbacks
@impl true
def init(%State{handler: handler} = opts) do
{:ok, _} = Tortoise.Events.register(opts.client_id, :status)
case Handler.execute(handler, :init) do
{:ok, %Handler{} = updated_handler} ->
{:ok, %State{opts | handler: updated_handler}}
end
end
@impl true
def terminate(reason, %State{handler: handler}) do
_ignored = Handler.execute(handler, {:terminate, reason})
:ok
end
@impl true
def handle_call(:info, _from, state) do
{:reply, state, state}
end
@impl true
def handle_cast({:incoming, <<package::binary>>}, state) do
package
|> Package.decode()
|> handle_package(state)
end
# allow for passing in already decoded packages into the controller,
# this allow us to test the controller without having to pass in
# binaries
def handle_cast({:incoming, %{:__META__ => _} = package}, state) do
handle_package(package, state)
end
def handle_cast({:ping, caller}, state) do
with {:ok, {transport, socket}} <- Connection.connection(state.client_id) do
time = System.monotonic_time(:microsecond)
apply(transport, :send, [socket, Package.encode(%Package.Pingreq{})])
ping = :queue.in({caller, time}, state.ping)
{:noreply, %State{state | ping: ping}}
else
{:error, :unknown_connection} ->
{:stop, :unknown_connection, state}
end
end
def handle_cast(
{:result, {Package.Subscribe, subacks}},
%State{handler: handler} = state
) do
case Handler.execute(handler, {:subscribe, subacks}) do
{:ok, updated_handler} ->
{:noreply, %State{state | handler: updated_handler}}
end
end
def handle_cast(
{:result, {Package.Unsubscribe, unsubacks}},
%State{handler: handler} = state
) do
case Handler.execute(handler, {:unsubscribe, unsubacks}) do
{:ok, updated_handler} ->
{:noreply, %State{state | handler: updated_handler}}
end
end
# an incoming publish with QoS=2 will get parked in the inflight
# manager process, which will onward it to the controller, making
# sure we will only dispatch it once to the publish-handler.
def handle_cast(
{:onward, %Package.Publish{qos: 2, dup: false} = publish},
%State{handler: handler} = state
) do
case Handler.execute(handler, {:publish, publish}) do
{:ok, updated_handler} ->
{:noreply, %State{state | handler: updated_handler}}
end
end
@impl true
def handle_info({:next_action, {:subscribe, topic, opts} = action}, state) do
{qos, opts} = Keyword.pop_first(opts, :qos, 0)
case Tortoise.Connection.subscribe(state.client_id, [{topic, qos}], opts) do
{:ok, ref} ->
updated_awaiting = Map.put_new(state.awaiting, ref, action)
{:noreply, %State{state | awaiting: updated_awaiting}}
end
end
def handle_info({:next_action, {:unsubscribe, topic} = action}, state) do
case Tortoise.Connection.unsubscribe(state.client_id, topic) do
{:ok, ref} ->
updated_awaiting = Map.put_new(state.awaiting, ref, action)
{:noreply, %State{state | awaiting: updated_awaiting}}
end
end
# connection changes
def handle_info(
{{Tortoise, client_id}, :status, same},
%State{client_id: client_id, status: same} = state
) do
{:noreply, state}
end
def handle_info(
{{Tortoise, client_id}, :status, new_status},
%State{client_id: client_id, handler: handler} = state
) do
case Handler.execute(handler, {:connection, new_status}) do
{:ok, updated_handler} ->
{:noreply, %State{state | handler: updated_handler, status: new_status}}
end
end
def handle_info({{Tortoise, client_id}, ref, result}, %{client_id: client_id} = state) do
case {result, Map.pop(state.awaiting, ref)} do
{_, {nil, _}} ->
Logger.warn("Unexpected async result")
{:noreply, state}
{:ok, {_action, updated_awaiting}} ->
{:noreply, %State{state | awaiting: updated_awaiting}}
end
end
# QoS LEVEL 0 ========================================================
# commands -----------------------------------------------------------
defp handle_package(
%Publish{qos: 0, dup: false} = publish,
%State{handler: handler} = state
) do
case Handler.execute(handler, {:publish, publish}) do
{:ok, updated_handler} ->
{:noreply, %State{state | handler: updated_handler}}
# handle stop
end
end
# QoS LEVEL 1 ========================================================
# commands -----------------------------------------------------------
defp handle_package(
%Publish{qos: 1} = publish,
%State{handler: handler} = state
) do
:ok = Inflight.track(state.client_id, {:incoming, publish})
case Handler.execute(handler, {:publish, publish}) do
{:ok, updated_handler} ->
{:noreply, %State{state | handler: updated_handler}}
end
end
# response -----------------------------------------------------------
defp handle_package(%Puback{} = puback, state) do
:ok = Inflight.update(state.client_id, {:received, puback})
{:noreply, state}
end
# QoS LEVEL 2 ========================================================
# commands -----------------------------------------------------------
defp handle_package(%Publish{qos: 2} = publish, %State{} = state) do
:ok = Inflight.track(state.client_id, {:incoming, publish})
{:noreply, state}
end
defp handle_package(%Pubrel{} = pubrel, state) do
:ok = Inflight.update(state.client_id, {:received, pubrel})
{:noreply, state}
end
# response -----------------------------------------------------------
defp handle_package(%Pubrec{} = pubrec, state) do
:ok = Inflight.update(state.client_id, {:received, pubrec})
{:noreply, state}
end
defp handle_package(%Pubcomp{} = pubcomp, state) do
:ok = Inflight.update(state.client_id, {:received, pubcomp})
{:noreply, state}
end
# SUBSCRIBING ========================================================
# command ------------------------------------------------------------
defp handle_package(%Subscribe{} = subscribe, state) do
# not a server! (yet)
{:stop, {:protocol_violation, {:unexpected_package_from_remote, subscribe}}, state}
end
# response -----------------------------------------------------------
defp handle_package(%Suback{} = suback, state) do
:ok = Inflight.update(state.client_id, {:received, suback})
{:noreply, state}
end
# UNSUBSCRIBING ======================================================
# command ------------------------------------------------------------
defp handle_package(%Unsubscribe{} = unsubscribe, state) do
# not a server
{:stop, {:protocol_violation, {:unexpected_package_from_remote, unsubscribe}}, state}
end
# response -----------------------------------------------------------
defp handle_package(%Unsuback{} = unsuback, state) do
:ok = Inflight.update(state.client_id, {:received, unsuback})
{:noreply, state}
end
# PING MESSAGES ======================================================
# command ------------------------------------------------------------
defp handle_package(%Pingresp{}, %State{ping: ping} = state)
when is_nil(ping) or ping == {[], []} do
{:noreply, state}
end
defp handle_package(%Pingresp{}, %State{ping: ping} = state) do
{{:value, {{caller, ref}, start_time}}, ping} = :queue.out(ping)
round_trip_time = System.monotonic_time(:microsecond) - start_time
send(caller, {Tortoise, {:ping_response, ref, round_trip_time}})
{:noreply, %State{state | ping: ping}}
end
# response -----------------------------------------------------------
defp handle_package(%Pingreq{} = pingreq, state) do
# not a server!
{:stop, {:protocol_violation, {:unexpected_package_from_remote, pingreq}}, state}
end
# CONNECTING =========================================================
# command ------------------------------------------------------------
defp handle_package(%Connect{} = connect, state) do
# not a server!
{:stop, {:protocol_violation, {:unexpected_package_from_remote, connect}}, state}
end
# response -----------------------------------------------------------
defp handle_package(%Connack{} = connack, state) do
# receiving a connack at this point would be a protocol violation
{:stop, {:protocol_violation, {:unexpected_package_from_remote, connack}}, state}
end
# DISCONNECTING ======================================================
# command ------------------------------------------------------------
defp handle_package(%Disconnect{} = disconnect, state) do
# This should be allowed when we implement MQTT 5. Remember there
# is a test that assert this as a protocol violation!
{:stop, {:protocol_violation, {:unexpected_package_from_remote, disconnect}}, state}
end
end
@@ -0,0 +1,378 @@
defmodule Tortoise.Connection.Inflight do
@moduledoc false
alias Tortoise.{Package, Connection}
alias Tortoise.Connection.Controller
alias Tortoise.Connection.Inflight.Track
use GenStateMachine
@enforce_keys [:client_id]
defstruct client_id: nil, pending: %{}, order: []
alias __MODULE__, as: State
# Client API
def start_link(opts) do
client_id = Keyword.fetch!(opts, :client_id)
GenStateMachine.start_link(__MODULE__, opts, name: via_name(client_id))
end
defp via_name(client_id) do
Tortoise.Registry.via_name(__MODULE__, client_id)
end
def stop(client_id) do
GenStateMachine.stop(via_name(client_id))
end
@doc false
def drain(client_id) do
GenStateMachine.call(via_name(client_id), :drain)
end
@doc false
def track(client_id, {:incoming, %Package.Publish{qos: qos} = publish})
when qos in 1..2 do
:ok = GenStateMachine.cast(via_name(client_id), {:incoming, publish})
end
def track(client_id, {:outgoing, package}) do
caller = {_, ref} = {self(), make_ref()}
case package do
%Package.Publish{qos: qos} when qos in 1..2 ->
:ok = GenStateMachine.cast(via_name(client_id), {:outgoing, caller, package})
{:ok, ref}
%Package.Subscribe{} ->
:ok = GenStateMachine.cast(via_name(client_id), {:outgoing, caller, package})
{:ok, ref}
%Package.Unsubscribe{} ->
:ok = GenStateMachine.cast(via_name(client_id), {:outgoing, caller, package})
{:ok, ref}
end
end
@doc false
def track_sync(client_id, {:outgoing, _} = command, timeout \\ :infinity) do
{:ok, ref} = track(client_id, command)
receive do
{{Tortoise, ^client_id}, ^ref, result} ->
result
after
timeout -> {:error, :timeout}
end
end
@doc false
def update(client_id, {_, %{__struct__: _, identifier: _identifier}} = event) do
:ok = GenStateMachine.cast(via_name(client_id), {:update, event})
end
@doc false
def reset(client_id) do
:ok = GenStateMachine.cast(via_name(client_id), :reset)
end
# Server callbacks
@impl true
def init(opts) do
client_id = Keyword.fetch!(opts, :client_id)
initial_data = %State{client_id: client_id}
next_actions = [
{:next_event, :internal, :post_init}
]
{:ok, :disconnected, initial_data, next_actions}
end
@impl true
def handle_event(:internal, :post_init, :disconnected, data) do
case Connection.connection(data.client_id, active: true) do
{:ok, {_transport, _socket} = connection} ->
{:ok, _} = Tortoise.Events.register(data.client_id, :status)
{:next_state, {:connected, connection}, data}
{:error, :timeout} ->
{:stop, :connection_timeout}
{:error, :unknown_connection} ->
{:stop, :unknown_connection}
end
end
# When we receive a new connection we will use that for our future
# transmissions.
def handle_event(
:info,
{{Tortoise, client_id}, :connection, connection},
_current_state,
%State{client_id: client_id, pending: pending} = data
) do
next_actions =
for identifier <- Enum.reverse(data.order) do
case Map.get(pending, identifier, :unknown) do
%Track{pending: [[{:dispatch, %Package.Publish{} = publish} | action] | pending]} =
track ->
publish = %Package.Publish{publish | dup: true}
track = %Track{track | pending: [[{:dispatch, publish} | action] | pending]}
{:next_event, :internal, {:execute, track}}
%Track{} = track ->
{:next_event, :internal, {:execute, track}}
end
end
{:next_state, {:connected, connection}, data, next_actions}
end
# Connection status events; when we go offline we should transition
# into the disconnected state. Everything else will get ignored.
def handle_event(
:info,
{{Tortoise, client_id}, :status, :down},
_current_state,
%State{client_id: client_id} = data
) do
{:next_state, :disconnected, data}
end
def handle_event(:info, {{Tortoise, _}, :status, _}, _, %State{}) do
:keep_state_and_data
end
# Create. Notice: we will only receive publish packages from the
# remote; everything else is something we initiate
def handle_event(
:cast,
{:incoming, %Package.Publish{dup: false} = package},
_state,
%State{pending: pending} = data
) do
track = Track.create(:positive, package)
data = %State{
data
| pending: Map.put_new(pending, track.identifier, track),
order: [track.identifier | data.order]
}
next_actions = [
{:next_event, :internal, {:onward_publish, package}},
{:next_event, :internal, {:execute, track}}
]
{:keep_state, data, next_actions}
end
# possible duplicate
def handle_event(:cast, {:incoming, _}, :draining, %State{}) do
:keep_state_and_data
end
def handle_event(
:cast,
{:incoming, %Package.Publish{identifier: identifier, dup: true} = publish},
_state,
%State{pending: pending} = data
) do
case Map.get(pending, identifier) do
nil ->
next_actions = [
{:next_event, :cast, {:incoming, %Package.Publish{publish | dup: false}}}
]
{:keep_state_and_data, next_actions}
%Track{polarity: :positive, status: [{:received, %{__struct__: Package.Publish}}]} ->
:keep_state_and_data
_otherwise ->
{:stop, :state_out_of_sync, data}
end
end
def handle_event(:cast, {:outgoing, {pid, ref}, _}, :draining, data) do
send(pid, {{Tortoise, data.client_id}, ref, {:error, :terminating}})
:keep_state_and_data
end
def handle_event(:cast, {:outgoing, caller, package}, _state, data) do
{:ok, package} = assign_identifier(package, data.pending)
track = Track.create({:negative, caller}, package)
next_actions = [
{:next_event, :internal, {:execute, track}}
]
data = %State{
data
| pending: Map.put_new(data.pending, track.identifier, track),
order: [track.identifier | data.order]
}
{:keep_state, data, next_actions}
end
# update
def handle_event(:cast, {:update, _}, :draining, _data) do
:keep_state_and_data
end
def handle_event(
:cast,
{:update, {_, %{identifier: identifier}} = update},
_state,
%State{pending: pending} = data
) do
with {:ok, track} <- Map.fetch(pending, identifier),
{:ok, track} <- Track.resolve(track, update) do
next_actions = [
{:next_event, :internal, {:execute, track}}
]
data = %State{
data
| pending: Map.put(pending, identifier, track),
order: [identifier | data.order -- [identifier]]
}
{:keep_state, data, next_actions}
else
:error ->
{:stop, {:protocol_violation, :unknown_identifier}, data}
{:error, reason} ->
{:stop, reason, data}
end
end
def handle_event(:cast, :reset, _, %State{pending: pending} = data) do
# cancel all currently outgoing messages
for {_, %Track{polarity: :negative, caller: {pid, ref}}} <- pending do
send(pid, {{Tortoise, data.client_id}, ref, {:error, :canceled}})
end
{:keep_state, %State{data | pending: %{}, order: []}}
end
# We trap the incoming QoS 2 packages in the inflight manager so we
# can make sure we will not onward them to the connection handler
# more than once.
def handle_event(
:internal,
{:onward_publish, %Package.Publish{qos: 2} = publish},
_,
%State{} = data
) do
:ok = Controller.handle_onward(data.client_id, publish)
:keep_state_and_data
end
# The other package types should not get onwarded to the controller
# handler
def handle_event(:internal, {:onward_publish, _}, _, %State{}) do
:keep_state_and_data
end
def handle_event(:internal, {:execute, _}, :draining, _) do
:keep_state_and_data
end
def handle_event(
:internal,
{:execute, %Track{pending: [[{:dispatch, package}, _] | _]} = track},
{:connected, {transport, socket}},
%State{} = data
) do
case apply(transport, :send, [socket, Package.encode(package)]) do
:ok ->
{:keep_state, handle_next(track, data)}
end
end
def handle_event(
:internal,
{:execute, %Track{pending: [[{:dispatch, _}, _] | _]}},
:disconnected,
%State{}
) do
# the dispatch will get re-queued when we regain the connection
:keep_state_and_data
end
def handle_event(
:internal,
{:execute, %Track{pending: [[{:respond, caller}, _] | _]} = track},
_state,
%State{client_id: client_id} = data
) do
case Track.result(track) do
{:ok, result} ->
:ok = Controller.handle_result(client_id, {caller, track.type, result})
{:keep_state, handle_next(track, data)}
end
end
def handle_event({:call, from}, :drain, {:connected, {transport, socket}}, %State{} = data) do
for {_, %Track{polarity: :negative, caller: {pid, ref}}} <- data.pending do
send(pid, {{Tortoise, data.client_id}, ref, {:error, :canceled}})
end
data = %State{data | pending: %{}, order: []}
disconnect = %Package.Disconnect{}
case apply(transport, :send, [socket, Package.encode(disconnect)]) do
:ok ->
:ok = transport.close(socket)
reply = {:reply, from, :ok}
{:next_state, :draining, data, reply}
end
end
# helpers ------------------------------------------------------------
defp handle_next(
%Track{pending: [[_, :cleanup]], identifier: identifier},
%State{pending: pending} = state
) do
order = state.order -- [identifier]
%State{state | pending: Map.delete(pending, identifier), order: order}
end
defp handle_next(_track, %State{} = state) do
state
end
# Assign a random identifier to the tracked package; this will make
# sure we pick a random number that is not in use
defp assign_identifier(%{identifier: nil} = package, pending) do
case :crypto.strong_rand_bytes(2) do
<<0, 0>> ->
# an identifier cannot be zero
assign_identifier(package, pending)
<<identifier::integer-size(16)>> ->
unless Map.has_key?(pending, identifier) do
{:ok, %{package | identifier: identifier}}
else
assign_identifier(package, pending)
end
end
end
# ...as such we should let the in-flight process assign identifiers,
# but the possibility to pass one in has been kept so we can make
# deterministic unit tests
defp assign_identifier(%{identifier: identifier} = package, pending)
when identifier in 0x0001..0xFFFF do
unless Map.has_key?(pending, identifier) do
{:ok, package}
else
{:error, {:identifier_already_in_use, identifier}}
end
end
end
@@ -0,0 +1,248 @@
defmodule Tortoise.Connection.Inflight.Track do
@moduledoc false
# A data structure implementing state machines tracking the state of a
# message in flight.
# Messages can have two polarities, positive and negative, describing
# what direction they are going. A positive polarity is messages
# coming from the server to the client (us), a negative polarity is
# messages send from the client (us) to the server.
# For now we care about tracking the state of a handful of message
# kinds: the publish control packages with a quality of service above
# 0 and subscribe and unsubscribe control packages. We do not track
# the in-flight state of a QoS 0 control packet because there is no
# state to track.
# For negative polarity we need to track the caller, which is the
# process that instantiated the publish control package. This process
# will wait for a message to get passed to it when the ownership of
# the control package has been transferred to the server. Messages
# with a positive polarity will get passed to the callback module
# attached to the Controller module, so in that case there will be no
# caller.
@type package :: Package.Publish | Package.Subscribe | Package.Unsubscribe
@type caller :: {pid(), reference()} | nil
@type polarity :: :positive | {:negative, caller()}
@type next_action :: {:dispatch | :expect, Tortoise.Encodable.t()}
@type status_update :: {:received | :dispatched, Tortoise.Encodable.t()}
@opaque t :: %__MODULE__{
polarity: :positive | :negative,
type: package,
caller: {pid(), reference()} | nil,
identifier: Tortoise.package_identifier(),
status: [status_update()],
pending: [next_action()]
}
@enforce_keys [:type, :identifier, :polarity, :pending]
defstruct type: nil,
polarity: nil,
caller: nil,
identifier: nil,
status: [],
pending: []
alias __MODULE__, as: State
alias Tortoise.Package
def next(%State{pending: [[next_action, resolution] | _]}) do
{next_action, resolution}
end
def resolve(%State{pending: [[action, :cleanup]]} = state, :cleanup) do
{:ok, %State{state | pending: [], status: [action | state.status]}}
end
def resolve(
%State{pending: [[action, {:received, %{__struct__: t, identifier: id}}] | rest]} = state,
{:received, %{__struct__: t, identifier: id}} = expected
) do
{:ok, %State{state | pending: rest, status: [expected, action | state.status]}}
end
# the value has previously been received; here we should stay where
# we are at and retry the transmission
def resolve(
%State{status: [{same, %{__struct__: t, identifier: id}} | _]} = state,
{same, %{__struct__: t, identifier: id}}
) do
{:ok, state}
end
def resolve(%State{pending: []} = state, :cleanup) do
{:ok, state}
end
def resolve(%State{}, {:received, package}) do
{:error, {:protocol_violation, {:unexpected_package_from_remote, package}}}
end
@type trackable :: Tortoise.Encodable
@doc """
Set up a data structure that will track the status of a control
packet
"""
# @todo, enable this when I've figured out what is wrong with this spec
# @spec create(polarity :: polarity(), package :: trackable()) :: __MODULE__.t()
def create(:positive, %Package.Publish{qos: 1, identifier: id} = publish) do
%State{
type: Package.Publish,
polarity: :positive,
identifier: id,
status: [{:received, publish}],
pending: [
[
{:dispatch, %Package.Puback{identifier: id}},
:cleanup
]
]
}
end
def create({:negative, {pid, ref}}, %Package.Publish{qos: 1, identifier: id} = publish)
when is_pid(pid) and is_reference(ref) do
%State{
type: Package.Publish,
polarity: :negative,
caller: {pid, ref},
identifier: id,
pending: [
[
{:dispatch, publish},
{:received, %Package.Puback{identifier: id}}
],
[
{:respond, {pid, ref}},
:cleanup
]
]
}
end
def create(:positive, %Package.Publish{identifier: id, qos: 2} = publish) do
%State{
type: Package.Publish,
polarity: :positive,
identifier: id,
status: [{:received, publish}],
pending: [
[
{:dispatch, %Package.Pubrec{identifier: id}},
{:received, %Package.Pubrel{identifier: id}}
],
[
{:dispatch, %Package.Pubcomp{identifier: id}},
:cleanup
]
]
}
end
def create({:negative, {pid, ref}}, %Package.Publish{identifier: id, qos: 2} = publish)
when is_pid(pid) and is_reference(ref) do
%State{
type: Package.Publish,
polarity: :negative,
caller: {pid, ref},
identifier: id,
pending: [
[
{:dispatch, publish},
{:received, %Package.Pubrec{identifier: id}}
],
[
{:dispatch, %Package.Pubrel{identifier: id}},
{:received, %Package.Pubcomp{identifier: id}}
],
[
{:respond, {pid, ref}},
:cleanup
]
]
}
end
# subscription
def create({:negative, {pid, ref}}, %Package.Subscribe{identifier: id} = subscribe)
when is_pid(pid) and is_reference(ref) do
%State{
type: Package.Subscribe,
polarity: :negative,
caller: {pid, ref},
identifier: id,
pending: [
[
{:dispatch, subscribe},
{:received, %Package.Suback{identifier: id}}
],
[
{:respond, {pid, ref}},
:cleanup
]
]
}
end
def create({:negative, {pid, ref}}, %Package.Unsubscribe{identifier: id} = unsubscribe)
when is_pid(pid) and is_reference(ref) do
%State{
type: Package.Unsubscribe,
polarity: :negative,
caller: {pid, ref},
identifier: id,
pending: [
[
{:dispatch, unsubscribe},
{:received, %Package.Unsuback{identifier: id}}
],
[
{:respond, {pid, ref}},
:cleanup
]
]
}
end
# calculate result
def result(%State{type: Package.Publish}) do
{:ok, :ok}
end
def result(%State{
type: Package.Unsubscribe,
status: [
{:received, _},
{:dispatch, %Package.Unsubscribe{topics: topics}} | _other
]
}) do
{:ok, topics}
end
def result(%State{
type: Package.Subscribe,
status: [
{:received, %Package.Suback{acks: acks}},
{:dispatch, %Package.Subscribe{topics: topics}} | _other
]
}) do
result =
List.zip([topics, acks])
|> Enum.reduce(%{error: [], warn: [], ok: []}, fn
{{topic, level}, {:ok, level}}, %{ok: oks} = acc ->
%{acc | ok: oks ++ [{topic, level}]}
{{topic, requested}, {:ok, actual}}, %{warn: warns} = acc ->
%{acc | warn: warns ++ [{topic, [requested: requested, accepted: actual]}]}
{{topic, level}, {:error, :access_denied}}, %{error: errors} = acc ->
%{acc | error: errors ++ [{:access_denied, {topic, level}}]}
end)
{:ok, result}
end
end
@@ -0,0 +1,197 @@
defmodule Tortoise.Connection.Receiver do
@moduledoc false
use GenStateMachine
alias Tortoise.Connection.Controller
alias Tortoise.Events
defstruct client_id: nil, transport: nil, socket: nil, buffer: <<>>
alias __MODULE__, as: State
def start_link(opts) do
client_id = Keyword.fetch!(opts, :client_id)
data = %State{client_id: client_id}
GenStateMachine.start_link(__MODULE__, data, name: via_name(client_id))
end
defp via_name(client_id) do
Tortoise.Registry.via_name(__MODULE__, client_id)
end
def child_spec(opts) do
%{
id: __MODULE__,
start: {__MODULE__, :start_link, [opts]},
type: :worker,
restart: :permanent,
shutdown: 500
}
end
def handle_socket(client_id, {transport, socket}) do
{:ok, pid} = GenStateMachine.call(via_name(client_id), {:handle_socket, transport, socket})
case transport.controlling_process(socket, pid) do
:ok ->
:ok
{:error, reason} when reason in [:closed, :einval] ->
# todo, this is an edge case, figure out what to do here
:ok
end
end
@impl true
def init(%State{} = data) do
{:ok, :disconnected, data}
end
@impl true
# receiving data on the network connection
def handle_event(:info, {transport, socket, tcp_data}, _, %{socket: socket} = data)
when transport in [:tcp, :ssl] do
next_actions = [
{:next_event, :internal, :activate_socket},
{:next_event, :internal, :consume_buffer}
]
new_data = %{data | buffer: <<data.buffer::binary, tcp_data::binary>>}
{:keep_state, new_data, next_actions}
end
# Dropped connection: tell the connection process that it should
# attempt to get a new network socket; unfortunately we cannot just
# monitor the socket port in the connection process as a transport
# method such as the SSL based one will pass an opaque data
# structure around instead of a port that can be monitored.
def handle_event(:info, {transport, socket}, _state, %{socket: socket} = data)
when transport in [:tcp_closed, :ssl_closed] do
# should we empty the buffer?
# communicate to the world that we have dropped the connection
:ok = Events.dispatch(data.client_id, :status, :down)
{:next_state, :disconnected, %{data | socket: nil}}
end
# activate network socket for incoming traffic
def handle_event(:internal, :activate_socket, _state_name, %State{transport: nil}) do
{:stop, :no_transport}
end
def handle_event(:internal, :activate_socket, _state_name, data) do
case data.transport.setopts(data.socket, active: :once) do
:ok ->
:keep_state_and_data
{:error, :einval} ->
# @todo consider if there could be a buffer we should drain at this point
{:next_state, :disconnected, data}
end
end
# consume buffer
def handle_event(:internal, :consume_buffer, _state_name, %{buffer: <<>>}) do
:keep_state_and_data
end
# consuming message
def handle_event(
:internal,
:consume_buffer,
{:connected, {:receiving_variable, length}},
%State{buffer: buffer} = data
)
when byte_size(buffer) >= length do
<<package::binary-size(length), rest::binary>> = buffer
next_state = {:connected, :receiving_fixed_header}
next_actions = [
{:next_event, :internal, {:emit, package}},
{:next_event, :internal, :consume_buffer}
]
new_data = %{data | buffer: rest}
{:next_state, next_state, new_data, next_actions}
end
def handle_event(:internal, :consume_buffer, {:connected, {:receiving_variable, _}}, _data) do
# await more bytes
:keep_state_and_data
end
# what should happen here?
def handle_event(:internal, :consume_buffer, :disconnected, _data) do
# we might be disconnected, but could we have data in the buffer still ?
# perhaps we should consume that
:keep_state_and_data
end
# receiving fixed header
def handle_event(:internal, :consume_buffer, {:connected, :receiving_fixed_header}, data) do
case parse_fixed_header(data.buffer) do
{:ok, length} ->
new_state = {:connected, {:receiving_variable, length}}
next_actions = [{:next_event, :internal, :consume_buffer}]
{:next_state, new_state, data, next_actions}
:cont ->
:keep_state_and_data
{:error, :invalid_header_length} ->
{:stop, {:protocol_violation, :invalid_header_length}}
end
end
def handle_event(:internal, {:emit, package}, _, data) do
:ok = Controller.handle_incoming(data.client_id, package)
:keep_state_and_data
end
def handle_event({:call, from}, {:handle_socket, transport, socket}, :disconnected, data) do
new_state = {:connected, :receiving_fixed_header}
next_actions = [
{:reply, from, {:ok, self()}},
{:next_event, :internal, :activate_socket},
{:next_event, :internal, :consume_buffer}
]
# better reset the buffer
new_data = %State{data | transport: transport, socket: socket, buffer: <<>>}
{:next_state, new_state, new_data, next_actions}
end
defp parse_fixed_header(<<_::8, 0::1, length::7, _::binary>>) do
{:ok, length + 2}
end
# 2 bytes
defp parse_fixed_header(<<_::8, 1::1, a::7, 0::1, b::7, _::binary>>) do
<<length::integer-size(14)>> = <<b::7, a::7>>
{:ok, length + 3}
end
# 3 bytes
defp parse_fixed_header(<<_::8, 1::1, a::7, 1::1, b::7, 0::1, c::7, _::binary>>) do
<<length::integer-size(21)>> = <<c::7, b::7, a::7>>
{:ok, length + 4}
end
# 4 bytes
defp parse_fixed_header(<<_::8, 1::1, a::7, 1::1, b::7, 1::1, c::7, 0::1, d::7, _::binary>>) do
<<length::integer-size(28)>> = <<d::7, c::7, b::7, a::7>>
{:ok, length + 5}
end
defp parse_fixed_header(header) when byte_size(header) > 5 do
{:error, :invalid_header_length}
end
defp parse_fixed_header(_) do
:cont
end
end
@@ -0,0 +1,27 @@
defmodule Tortoise.Connection.Supervisor do
@moduledoc false
use Supervisor
alias Tortoise.Connection.{Receiver, Controller, Inflight}
def start_link(opts) do
client_id = Keyword.fetch!(opts, :client_id)
Supervisor.start_link(__MODULE__, opts, name: via_name(client_id))
end
defp via_name(client_id) do
Tortoise.Registry.via_name(__MODULE__, client_id)
end
@impl true
def init(opts) do
children = [
{Inflight, Keyword.take(opts, [:client_id])},
{Receiver, Keyword.take(opts, [:client_id])},
{Controller, Keyword.take(opts, [:client_id, :handler])}
]
Supervisor.init(children, strategy: :rest_for_one)
end
end
@@ -0,0 +1,47 @@
defprotocol Tortoise.Decodable do
@moduledoc false
def decode(data)
end
defimpl Tortoise.Decodable, for: BitString do
alias Tortoise.Package.{
Connect,
Connack,
Publish,
Puback,
Pubrec,
Pubrel,
Pubcomp,
Subscribe,
Suback,
Unsubscribe,
Unsuback,
Pingreq,
Pingresp,
Disconnect
}
def decode(<<1::4, _::4, _::binary>> = data), do: Connect.decode(data)
def decode(<<2::4, _::4, _::binary>> = data), do: Connack.decode(data)
def decode(<<3::4, _::4, _::binary>> = data), do: Publish.decode(data)
def decode(<<4::4, _::4, _::binary>> = data), do: Puback.decode(data)
def decode(<<5::4, _::4, _::binary>> = data), do: Pubrec.decode(data)
def decode(<<6::4, _::4, _::binary>> = data), do: Pubrel.decode(data)
def decode(<<7::4, _::4, _::binary>> = data), do: Pubcomp.decode(data)
def decode(<<8::4, _::4, _::binary>> = data), do: Subscribe.decode(data)
def decode(<<9::4, _::4, _::binary>> = data), do: Suback.decode(data)
def decode(<<10::4, _::4, _::binary>> = data), do: Unsubscribe.decode(data)
def decode(<<11::4, _::4, _::binary>> = data), do: Unsuback.decode(data)
def decode(<<12::4, _::4, _::binary>> = data), do: Pingreq.decode(data)
def decode(<<13::4, _::4, _::binary>> = data), do: Pingresp.decode(data)
def decode(<<14::4, _::4, _::binary>> = data), do: Disconnect.decode(data)
end
defimpl Tortoise.Decodable, for: List do
def decode(data) do
data
|> IO.iodata_to_binary()
|> Tortoise.Decodable.decode()
end
end
@@ -0,0 +1,6 @@
defprotocol Tortoise.Encodable do
@moduledoc false
@spec encode(t) :: iodata()
def encode(package)
end
@@ -0,0 +1,70 @@
defmodule Tortoise.Events do
@moduledoc """
A PubSub exposing various system events from a Tortoise
connection. This allows the user to integrate with custom metrics
and logging solutions.
Please read the documentation for `Tortoise.Events.register/2` for
information on how to subscribe to events, and
`Tortoise.Events.unregister/2` for how to unsubscribe.
"""
@types [:connection, :status, :ping_response]
@doc """
Subscribe to messages on the client with the client id `client_id`
of the type `type`.
When a message of the subscribed type is dispatched it will end up
in the mailbox of the process that placed the subscription. The
received message will have the format:
{{Tortoise, client_id}, type, value}
Making it possible to pattern match on multiple message types on
multiple clients. The value depends on the message type.
Possible message types are:
- `:status` dispatched when the connection of a client changes
status. The value will be `:up` when the client goes online, and
`:down` when it goes offline.
- `:ping_response` dispatched when the connection receive a
response from a keep alive message. The value is the round trip
time in milliseconds, and can be used to track the latency over
time.
Other message types exist, but unless they are mentioned in the
possible message types above they should be considered for internal
use only.
It is possible to listen on all events for a given type by
specifying `:_` as the `client_id`.
"""
@spec register(Tortoise.client_id(), atom()) :: {:ok, pid()} | no_return()
def register(client_id, type) when type in @types do
{:ok, _pid} = Registry.register(__MODULE__, type, client_id)
end
@doc """
Unsubscribe from messages of `type` from `client_id`. This is the
reverse of `Tortoise.Events.register/2`.
"""
@spec unregister(Tortoise.client_id(), atom()) :: :ok | no_return()
def unregister(client_id, type) when type in @types do
:ok = Registry.unregister_match(__MODULE__, type, client_id)
end
@doc false
@spec dispatch(Tortoise.client_id(), type :: atom(), value :: term()) :: :ok
def dispatch(client_id, type, value) when type in @types do
:ok =
Registry.dispatch(__MODULE__, type, fn subscribers ->
for {pid, filter} <- subscribers,
filter == client_id or filter == :_ do
Kernel.send(pid, {{Tortoise, client_id}, type, value})
end
end)
end
end
@@ -0,0 +1,413 @@
defmodule Tortoise.Handler do
@moduledoc """
User defined callback module for handling connection life cycle events.
`Tortoise.Handler` defines a behaviour which can be given to a
`Tortoise.Connection`. This allow the user to implement
functionality for events happening in the life cycle of the
connection, the provided callbacks are:
- `init/1` and `terminate/2` are called when the connection is
started and stopped. Parts that are setup in `init/1` can be
torn down in `terminate/2`. Notice that the connection process
is not terminated if the connection to the broker is lost;
`Tortoise` will attempt to reconnect, and events that should
happen when the connection goes offline should be set up using
the `connection/3` callback.
- `connection/3` is called when the connection is `:up` or
`:down`, allowing for functionality to be run when the
connection state changes.
- `subscription/3` is called when a topic filter subscription
changes status, so this callback can be used to control the
life-cycle of a subscription, allowing us to implement custom
behavior for when the subscription is accepted, declined, as
well as unsubscribed.
- `handle_message/3` is run when the client receive a message on
one of the subscribed topic filters.
Because the callback-module will run inside the connection
controller process, which also handles the routing of protocol
messages (such as publish acknowledge messages) it is important that
the callbacks do not call functions that will block the process,
especially for clients that subscribe to topics with heavy
traffic.
Technically it would be possible to run the callback module in a
different process than the controller process, but it has been
decided to keep it on the controller as we otherwise would have to
copy every message evaluated by the connection to another
process. It is much better to let the end-user handle the
dispatching to other parts of the system while we are evaluating
what to do with the process anyways with the caveats:
- The callbacks should not block the controller
- it is not possible to call the (un)subscribe and publish
functions from within a callback as they will block the
controller.
While it is not possible to subscribe and unsubscribe in the handler
process using the `Tortoise.subscribe/3` and
`Tortoise.unsubscribe/3` it is possible to make changes to the
subscription list via `:gen_statem` inspired *next_actions*.
## Next actions
In some situations one would like to subscribe or unsubscribe a
filter topic when a certain event happens on the client. The
functions for interacting with the MQTT broker, defined on the
`Tortoise`-module are for the most part blocking operations, or
require the user to peek into the process mailbox to fetch the
result of the operation. To allow for changes in the subscriptions
one can define a set of next actions that should happen as part of
the return value to the `handle_message/3`, `subscription/3`, and
`connection/3` callbacks by returning a `{:ok, state, next_actions}`
where `next_actions` is a list of commands of:
- `{:subscribe, topic_filter, qos: qos, timeout: 5000}` where
`topic_filter` is a binary containing a valid MQTT topic filter,
and `qos` is the desired quality of service (0..2). The timeout
is the amount of time in milliseconds we are willing to wait for
a response to the request.
- `{:unsubscribe, topic_filter}` where `topic_filter` is a binary
containing the name of the subscription we want to unsubscribe
from.
If we want to unsubscribe from the current topic when we receive a
message on it we could write a `handle_message/3` as follows:
def handle_message(topic, _payload, state) do
topic = Enum.join(topic, "/")
next_actions = [{:unsubscribe, topic}]
{:ok, state, next_actions}
end
Note that the `topic` is received as a list of topic levels, and
that the next actions has to be a list, even if there is only one
next action; multiple actions can be given at once. Read more about
this in the `handle_message/3` documentation.
"""
alias Tortoise.Package
@typedoc """
Data structure describing the user defined callback handler
The data structure describe the current state as well as its initial
arguments and the module driving the handler. This allow Tortoise to
restart the handler if needed be.
"""
@type t :: %__MODULE__{
module: module(),
initial_args: term(),
state: term()
}
@enforce_keys [:module, :initial_args]
defstruct module: nil, state: nil, initial_args: []
# Helper for building a Handler struct so we can keep it as an
# opaque type in the system.
@doc false
@spec new({module(), args :: term()}) :: t
def new({module, args}) when is_atom(module) and is_list(args) do
%__MODULE__{module: module, initial_args: args}
end
# identity
def new(%__MODULE__{} = handler), do: handler
defmacro __using__(_opts) do
quote location: :keep do
@behaviour Tortoise.Handler
@impl true
def init(state) do
{:ok, state}
end
@impl true
def terminate(_reason, _state) do
:ok
end
@impl true
def connection(_status, state) do
{:ok, state}
end
@impl true
def subscription(_status, _topic_filter, state) do
{:ok, state}
end
@impl true
def handle_message(_topic, _payload, state) do
{:ok, state}
end
defoverridable Tortoise.Handler
end
end
@typedoc """
Action to perform before reentering the execution loop.
The supported next actions are:
- Tell the connection process to subscribe to a topic filter
- Tell the connection process to unsubscribe from a topic filter
More next actions might be supported in the future.
"""
@type next_action() ::
{:subscribe, Tortoise.topic_filter(), [{:qos, Tortoise.qos()} | {:timeout, timeout()}]}
| {:unsubscribe, Tortoise.topic_filter()}
@doc """
Invoked when the connection is started.
`args` is the argument passed in from the connection configuration.
Returning `{:ok, state}` will let the MQTT connection receive data
from the MQTT broker, and the value contained in `state` will be
used as the process state.
"""
@callback init(args :: term()) :: {:ok, state}
when state: any()
@doc """
Invoked when the connection status changes.
`status` is one of `:up` or `:down`, where up means we have an open
connection to the MQTT broker, and down means the connection is
temporary down. The connection process will attempt to reestablish
the connection.
Returning `{:ok, new_state}` will set the state for later
invocations.
Returning `{:ok, new_state, next_actions}`, where `next_actions` is
a list of next actions such as `{:unsubscribe, "foo/bar"}` will
result in the state being returned and the next actions performed.
"""
@callback connection(status, state :: term()) ::
{:ok, new_state}
| {:ok, new_state, [next_action()]}
when status: :up | :down,
new_state: term()
@doc """
Invoked when the subscription of a topic filter changes status.
The `status` of a subscription can be one of:
- `:up`, triggered when the subscription has been accepted by the
MQTT broker with the requested quality of service
- `{:warn, [requested: req_qos, accepted: qos]}`, triggered when
the subscription is accepted by the MQTT broker, but with a
different quality of service `qos` than the one requested
`req_qos`
- `{:error, reason}`, triggered when the subscription is rejected
with the reason `reason` such as `:access_denied`
- `:down`, triggered when the subscription of the given topic
filter has been successfully acknowledged as unsubscribed by the
MQTT broker
The `topic_filter` is the topic filter in question, and the `state`
is the internal state being passed through transitions.
Returning `{:ok, new_state}` will set the state for later
invocations.
Returning `{:ok, new_state, next_actions}`, where `next_actions` is
a list of next actions such as `{:unsubscribe, "foo/bar"}` will
result in the state being returned and the next actions performed.
"""
@callback subscription(status, topic_filter, state :: term) ::
{:ok, new_state}
| {:ok, new_state, [next_action()]}
when status:
:up
| :down
| {:warn, [requested: Tortoise.qos(), accepted: Tortoise.qos()]}
| {:error, term()},
topic_filter: Tortoise.topic_filter(),
new_state: term
@doc """
Invoked when messages are published to subscribed topics.
The `topic` comes in the form of a list of binaries, making it
possible to pattern match on the topic levels of the retrieved
message, store the individual topic levels as variables and use it
in the function body.
`Payload` is a binary. MQTT 3.1.1 does not specify any format of the
payload, so it has to be decoded and validated depending on the
needs of the application.
In an example where we are already subscribed to the topic filter
`room/+/temp` and want to dispatch the received messages to a
`Temperature` application we could set up our `handle_message` as
such:
def handle_message(["room", room, "temp"], payload, state) do
:ok = Temperature.record(room, payload)
{:ok, state}
end
Notice; the `handle_message/3`-callback run inside the connection
controller process, so for handlers that are subscribing to topics
with heavy traffic should do as little as possible in the callback
handler and dispatch to other parts of the application using
non-blocking calls.
Returning `{:ok, new_state}` will reenter the loop and set the state
for later invocations.
Returning `{:ok, new_state, next_actions}`, where `next_actions` is
a list of next actions such as `{:unsubscribe, "foo/bar"}` will
reenter the loop and perform the listed actions.
"""
@callback handle_message(topic_levels, payload, state :: term()) ::
{:ok, new_state}
| {:ok, new_state, [next_action()]}
when new_state: term(),
topic_levels: [String.t()],
payload: Tortoise.payload()
@doc """
Invoked when the connection process is about to exit.
If anything is setup during the `init/1` callback it should get
cleaned up during the `terminate/2` callback.
"""
@callback terminate(reason, state :: term) :: ignored
when reason: :normal | :shutdown | {:shutdown, term()},
ignored: term()
@doc false
@spec execute(t, action) :: :ok | {:ok, t} | {:error, {:invalid_next_action, term()}}
when action:
:init
| {:subscribe, [term()]}
| {:unsubscribe, [term()]}
| {:publish, Tortoise.Package.Publish.t()}
| {:connection, :up | :down}
| {:terminate, reason :: term()}
def execute(handler, :init) do
case apply(handler.module, :init, [handler.initial_args]) do
{:ok, initial_state} ->
{:ok, %__MODULE__{handler | state: initial_state}}
end
end
def execute(handler, {:connection, status}) do
handler.module
|> apply(:connection, [status, handler.state])
|> handle_result(handler)
end
def execute(handler, {:publish, %Package.Publish{} = publish}) do
topic_list = String.split(publish.topic, "/")
handler.module
|> apply(:handle_message, [topic_list, publish.payload, handler.state])
|> handle_result(handler)
end
def execute(handler, {:unsubscribe, unsubacks}) do
Enum.reduce(unsubacks, {:ok, handler}, fn topic_filter, {:ok, handler} ->
handler.module
|> apply(:subscription, [:down, topic_filter, handler.state])
|> handle_result(handler)
# _, {:stop, acc} ->
# {:stop, acc}
end)
end
def execute(handler, {:subscribe, subacks}) do
subacks
|> flatten_subacks()
|> Enum.reduce({:ok, handler}, fn {op, topic_filter}, {:ok, handler} ->
handler.module
|> apply(:subscription, [op, topic_filter, handler.state])
|> handle_result(handler)
# _, {:stop, acc} ->
# {:stop, acc}
end)
end
def execute(handler, {:terminate, reason}) do
_ignored = apply(handler.module, :terminate, [reason, handler.state])
:ok
end
# Subacks will come in a map with three keys in the form of tuples
# where the fist element is one of `:ok`, `:warn`, or `:error`. This
# is done to make it easy to pattern match in other parts of the
# system, and error out early if the result set contain errors. In
# this part of the system it is more convenient to transform the
# data to a flat list containing tuples of `{operation, data}` so we
# can reduce the handler state to collect the possible next actions,
# and pass through if there is an :error or :disconnect return.
defp flatten_subacks(subacks) do
Enum.reduce(subacks, [], fn
{_, []}, acc ->
acc
{:ok, entries}, acc ->
for {topic_filter, _qos} <- entries do
{:up, topic_filter}
end ++ acc
{:warn, entries}, acc ->
for {topic_filter, warning} <- entries do
{{:warn, warning}, topic_filter}
end ++ acc
{:error, entries}, acc ->
for {reason, {topic_filter, _qos}} <- entries do
{{:error, reason}, topic_filter}
end ++ acc
end)
end
# handle the user defined return from the callback
defp handle_result({:ok, updated_state}, handler) do
{:ok, %__MODULE__{handler | state: updated_state}}
end
defp handle_result({:ok, updated_state, next_actions}, handler)
when is_list(next_actions) do
case Enum.split_with(next_actions, &valid_next_action?/1) do
{next_actions, []} ->
# send the next actions to the process mailbox. Notice that
# this code is run in the context of the connection controller
for action <- next_actions, do: send(self(), {:next_action, action})
{:ok, %__MODULE__{handler | state: updated_state}}
{_, errors} ->
{:error, {:invalid_next_action, errors}}
end
end
defp valid_next_action?({:subscribe, topic, opts}) do
is_binary(topic) and is_list(opts)
end
defp valid_next_action?({:unsubscribe, topic}) do
is_binary(topic)
end
defp valid_next_action?(_otherwise), do: false
end
@@ -0,0 +1,5 @@
defmodule Tortoise.Handler.Default do
@moduledoc false
use Tortoise.Handler
end
@@ -0,0 +1,60 @@
defmodule Tortoise.Handler.Logger do
@moduledoc false
require Logger
use Tortoise.Handler
defstruct []
alias __MODULE__, as: State
def init(_opts) do
Logger.info("Initializing handler")
{:ok, %State{}}
end
def connection(:up, state) do
Logger.info("Connection has been established")
{:ok, state}
end
def connection(:down, state) do
Logger.warn("Connection has been dropped")
{:ok, state}
end
def connection(:terminating, state) do
Logger.warn("Connection is terminating")
{:ok, state}
end
def subscription(:up, topic, state) do
Logger.info("Subscribed to #{topic}")
{:ok, state}
end
def subscription({:warn, [requested: req, accepted: qos]}, topic, state) do
Logger.warn("Subscribed to #{topic}; requested #{req} but got accepted with QoS #{qos}")
{:ok, state}
end
def subscription({:error, reason}, topic, state) do
Logger.error("Error subscribing to #{topic}; #{inspect(reason)}")
{:ok, state}
end
def subscription(:down, topic, state) do
Logger.info("Unsubscribed from #{topic}")
{:ok, state}
end
def handle_message(topic, publish, state) do
Logger.info("#{Enum.join(topic, "/")} #{inspect(publish)}")
{:ok, state}
end
def terminate(reason, _state) do
Logger.warn("Client has been terminated with reason: #{inspect(reason)}")
:ok
end
end
@@ -0,0 +1,43 @@
defmodule Tortoise.Package do
@moduledoc false
alias Tortoise.Package
@opaque message ::
Package.Connect.t()
| Package.Connack.t()
| Package.Publish.t()
| Package.Puback.t()
| Package.Pubrec.t()
| Package.Pubrel.t()
| Package.Pubcomp.t()
| Package.Subscribe.t()
| Package.Suback.t()
| Package.Unsubscribe.t()
| Package.Unsuback.t()
| Package.Pingreq.t()
| Package.Pingresp.t()
| Package.Disconnect.t()
defdelegate encode(data), to: Tortoise.Encodable
defdelegate decode(data), to: Tortoise.Decodable
@doc false
def length_encode(data) do
length_prefix = <<byte_size(data)::big-integer-size(16)>>
[length_prefix, data]
end
@doc false
def variable_length_encode(data) when is_list(data) do
length_prefix = data |> IO.iodata_length() |> remaining_length()
length_prefix ++ data
end
@highbit 0b10000000
defp remaining_length(n) when n < @highbit, do: [<<0::1, n::7>>]
defp remaining_length(n) do
[<<1::1, rem(n, @highbit)::7>>] ++ remaining_length(div(n, @highbit))
end
end
@@ -0,0 +1,69 @@
defmodule Tortoise.Package.Connack do
@moduledoc false
@opcode 2
alias Tortoise.Package
@type status :: :accepted | {:refused, refusal_reasons()}
@type refusal_reasons ::
:unacceptable_protocol_version
| :identifier_rejected
| :server_unavailable
| :bad_user_name_or_password
| :not_authorized
@opaque t :: %__MODULE__{
__META__: Package.Meta.t(),
session_present: boolean(),
status: status() | nil
}
@enforce_keys [:status]
defstruct __META__: %Package.Meta{opcode: @opcode, flags: 0},
session_present: false,
status: nil
@spec decode(<<_::32>>) :: t
def decode(<<@opcode::4, 0::4, 2, 0::7, session_present::1, return_code::8>>) do
%__MODULE__{
session_present: session_present == 1,
status: coerce_return_code(return_code)
}
end
defp coerce_return_code(return_code) do
case return_code do
0x00 -> :accepted
0x01 -> {:refused, :unacceptable_protocol_version}
0x02 -> {:refused, :identifier_rejected}
0x03 -> {:refused, :server_unavailable}
0x04 -> {:refused, :bad_user_name_or_password}
0x05 -> {:refused, :not_authorized}
end
end
defimpl Tortoise.Encodable do
def encode(%Package.Connack{session_present: session_present, status: status} = t)
when status != nil do
[
Package.Meta.encode(t.__META__),
<<2, 0::7, flag(session_present)::1, to_return_code(status)::8>>
]
end
defp to_return_code(:accepted), do: 0x00
defp to_return_code({:refused, reason}) do
case reason do
:unacceptable_protocol_version -> 0x01
:identifier_rejected -> 0x02
:server_unavailable -> 0x03
:bad_user_name_or_password -> 0x04
:not_authorized -> 0x05
end
end
defp flag(f) when f in [0, nil, false], do: 0
defp flag(_), do: 1
end
end
@@ -0,0 +1,161 @@
defmodule Tortoise.Package.Connect do
@moduledoc false
@opcode 1
alias Tortoise.Package
@opaque t :: %__MODULE__{
__META__: Package.Meta.t(),
protocol: binary(),
protocol_version: non_neg_integer(),
user_name: binary() | nil,
password: binary() | nil,
clean_session: boolean(),
keep_alive: non_neg_integer(),
client_id: Tortoise.client_id(),
will: Package.Publish.t() | nil
}
@enforce_keys [:client_id]
defstruct __META__: %Package.Meta{opcode: @opcode},
protocol: "MQTT",
protocol_version: 0b00000100,
user_name: nil,
password: nil,
clean_session: true,
keep_alive: 60,
client_id: nil,
will: nil
@spec decode(binary()) :: t
def decode(<<@opcode::4, 0::4, variable::binary>>) do
<<4::big-integer-size(16), "MQTT", 4::8, user_name::1, password::1, will_retain::1,
will_qos::2, will::1, clean_session::1, 0::1, keep_alive::big-integer-size(16),
package::binary>> = drop_length(variable)
options =
[
client_id: 1,
will_topic: will,
will_payload: will,
user_name: user_name,
password: password
]
|> Enum.filter(fn {_, present} -> present == 1 end)
|> Enum.map(fn {value, 1} -> value end)
|> Enum.zip(decode_length_prefixed(package))
%__MODULE__{
client_id: options[:client_id],
user_name: options[:user_name],
password: options[:password],
will:
if will == 1 do
%Package.Publish{
topic: options[:will_topic],
payload: nullify(options[:will_payload]),
qos: will_qos,
retain: will_retain == 1
}
end,
clean_session: clean_session == 1,
keep_alive: keep_alive
}
end
defp nullify(""), do: nil
defp nullify(payload), do: payload
defp drop_length(payload) do
case payload do
<<0::1, _::7, r::binary>> -> r
<<1::1, _::7, 0::1, _::7, r::binary>> -> r
<<1::1, _::7, 1::1, _::7, 0::1, _::7, r::binary>> -> r
<<1::1, _::7, 1::1, _::7, 1::1, _::7, 0::1, _::7, r::binary>> -> r
end
end
defp decode_length_prefixed(<<>>), do: []
defp decode_length_prefixed(<<length::big-integer-size(16), payload::binary>>) do
<<item::binary-size(length), rest::binary>> = payload
[item] ++ decode_length_prefixed(rest)
end
defimpl Tortoise.Encodable do
def encode(%Package.Connect{client_id: client_id} = t)
when is_binary(client_id) do
[
Package.Meta.encode(t.__META__),
Package.variable_length_encode([
protocol_header(t),
connection_flags(t),
keep_alive(t),
payload(t)
])
]
end
def encode(%Package.Connect{client_id: client_id} = t)
when is_atom(client_id) do
encode(%Package.Connect{t | client_id: Atom.to_string(client_id)})
end
defp protocol_header(%{protocol: protocol, protocol_version: version}) do
[Package.length_encode(protocol), version]
end
defp connection_flags(%{will: nil} = f) do
<<
flag(f.user_name)::integer-size(1),
flag(f.password)::integer-size(1),
# will retain
flag(0)::integer-size(1),
# will qos
0::integer-size(2),
# will flag
flag(0)::integer-size(1),
flag(f.clean_session)::integer-size(1),
# reserved bit
0::1
>>
end
defp connection_flags(%{will: %Package.Publish{}} = f) do
<<
flag(f.user_name)::integer-size(1),
flag(f.password)::integer-size(1),
flag(f.will.retain)::integer-size(1),
f.will.qos::integer-size(2),
flag(f.will.topic)::integer-size(1),
flag(f.clean_session)::integer-size(1),
# reserved bit
0::1
>>
end
defp keep_alive(f) do
<<f.keep_alive::big-integer-size(16)>>
end
defp payload(%{will: nil} = f) do
[f.client_id, f.user_name, f.password]
|> Enum.filter(&is_binary/1)
|> Enum.map(&Package.length_encode/1)
end
defp payload(f) do
will_payload = encode_payload(f.will.payload)
[f.client_id, f.will.topic, will_payload, f.user_name, f.password]
|> Enum.filter(&is_binary/1)
|> Enum.map(&Package.length_encode/1)
end
defp encode_payload(nil), do: ""
defp encode_payload(payload), do: payload
defp flag(f) when f in [0, nil, false], do: 0
defp flag(_), do: 1
end
end
@@ -0,0 +1,22 @@
defmodule Tortoise.Package.Disconnect do
@moduledoc false
@opcode 14
alias Tortoise.Package
@opaque t :: %__MODULE__{
__META__: Package.Meta.t()
}
defstruct __META__: %Package.Meta{opcode: @opcode, flags: 0}
@spec decode(<<_::16>>) :: t
def decode(<<@opcode::4, 0::4, 0>>), do: %__MODULE__{}
# Protocols ----------------------------------------------------------
defimpl Tortoise.Encodable do
def encode(%Package.Disconnect{} = t) do
[Package.Meta.encode(t.__META__), 0]
end
end
end
@@ -0,0 +1,14 @@
defmodule Tortoise.Package.Meta do
@moduledoc false
@opaque t() :: %__MODULE__{
opcode: 0 | 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8 | 9 | 10 | 11 | 12 | 13 | 14,
flags: non_neg_integer()
}
@enforce_keys [:opcode]
defstruct opcode: 0, flags: 0
def encode(meta) do
<<meta.opcode::4, meta.flags::4>>
end
end
@@ -0,0 +1,24 @@
defmodule Tortoise.Package.Pingreq do
@moduledoc false
@opcode 12
alias Tortoise.Package
@opaque t :: %__MODULE__{
__META__: Package.Meta.t()
}
defstruct __META__: %Package.Meta{opcode: @opcode, flags: 0}
@spec decode(<<_::16>>) :: t
def decode(<<@opcode::4, 0::4, 0>>) do
%__MODULE__{}
end
# Protocols ----------------------------------------------------------
defimpl Tortoise.Encodable do
def encode(%Package.Pingreq{} = t) do
[Package.Meta.encode(t.__META__), 0]
end
end
end
@@ -0,0 +1,24 @@
defmodule Tortoise.Package.Pingresp do
@moduledoc false
@opcode 13
alias Tortoise.Package
@opaque t :: %__MODULE__{
__META__: Package.Meta.t()
}
defstruct __META__: %Package.Meta{opcode: @opcode, flags: 0}
@spec decode(<<_::16>>) :: t
def decode(<<@opcode::4, 0::4, 0>>) do
%__MODULE__{}
end
# Protocols ----------------------------------------------------------
defimpl Tortoise.Encodable do
def encode(%Package.Pingresp{} = t) do
[Package.Meta.encode(t.__META__), 0]
end
end
end
@@ -0,0 +1,29 @@
defmodule Tortoise.Package.Puback do
@moduledoc false
@opcode 4
alias Tortoise.Package
@opaque t :: %__MODULE__{
__META__: Package.Meta.t(),
identifier: Tortoise.package_identifier()
}
@enforce_keys [:identifier]
defstruct __META__: %Package.Meta{opcode: @opcode, flags: 0b0000},
identifier: nil
@spec decode(<<_::32>>) :: t
def decode(<<@opcode::4, 0::4, 2, identifier::big-integer-size(16)>>)
when identifier in 0x0001..0xFFFF do
%__MODULE__{identifier: identifier}
end
# Protocols ----------------------------------------------------------
defimpl Tortoise.Encodable do
def encode(%Package.Puback{identifier: identifier} = t)
when identifier in 0x0001..0xFFFF do
[Package.Meta.encode(t.__META__), <<2, identifier::big-integer-size(16)>>]
end
end
end
@@ -0,0 +1,28 @@
defmodule Tortoise.Package.Pubcomp do
@moduledoc false
@opcode 7
alias Tortoise.Package
@opaque t :: %__MODULE__{
__META__: Package.Meta.t(),
identifier: Tortoise.package_identifier()
}
defstruct __META__: %Package.Meta{opcode: @opcode, flags: 0},
identifier: nil
@spec decode(<<_::32>>) :: t
def decode(<<@opcode::4, 0::4, 2, identifier::big-integer-size(16)>>)
when identifier in 0x0001..0xFFFF do
%__MODULE__{identifier: identifier}
end
# Protocols ----------------------------------------------------------
defimpl Tortoise.Encodable do
def encode(%Package.Pubcomp{identifier: identifier} = t)
when identifier in 0x0001..0xFFFF do
[Package.Meta.encode(t.__META__), <<2, t.identifier::big-integer-size(16)>>]
end
end
end
@@ -0,0 +1,114 @@
defmodule Tortoise.Package.Publish do
@moduledoc false
@opcode 3
alias Tortoise.Package
@type t :: %__MODULE__{
__META__: Package.Meta.t(),
topic: Tortoise.topic() | nil,
qos: Tortoise.qos(),
payload: Tortoise.payload(),
identifier: Tortoise.package_identifier(),
dup: boolean(),
retain: boolean()
}
defstruct __META__: %Package.Meta{opcode: @opcode, flags: 0},
identifier: nil,
topic: nil,
payload: nil,
qos: 0,
dup: false,
retain: false
@spec decode(binary()) :: t
def decode(<<@opcode::4, 0::1, 0::2, retain::1, length_prefixed_payload::binary>>) do
payload = drop_length_prefix(length_prefixed_payload)
{topic, payload} = decode_message(payload)
%__MODULE__{
qos: 0,
identifier: nil,
dup: false,
retain: retain == 1,
topic: topic,
payload: payload
}
end
def decode(
<<@opcode::4, dup::1, qos::integer-size(2), retain::1, length_prefixed_payload::binary>>
) do
payload = drop_length_prefix(length_prefixed_payload)
{topic, identifier, payload} = decode_message_with_id(payload)
%__MODULE__{
qos: qos,
identifier: identifier,
dup: dup == 1,
retain: retain == 1,
topic: topic,
payload: payload
}
end
defp drop_length_prefix(payload) do
case payload do
<<0::1, _::7, r::binary>> -> r
<<1::1, _::7, 0::1, _::7, r::binary>> -> r
<<1::1, _::7, 1::1, _::7, 0::1, _::7, r::binary>> -> r
<<1::1, _::7, 1::1, _::7, 1::1, _::7, 0::1, _::7, r::binary>> -> r
end
end
defp decode_message(<<topic_length::big-integer-size(16), msg::binary>>) do
<<topic::binary-size(topic_length), payload::binary>> = msg
{topic, nullify(payload)}
end
defp decode_message_with_id(<<topic_length::big-integer-size(16), msg::binary>>) do
<<topic::binary-size(topic_length), identifier::big-integer-size(16), payload::binary>> = msg
{topic, identifier, nullify(payload)}
end
defp nullify(""), do: nil
defp nullify(payload), do: payload
# Protocols ----------------------------------------------------------
defimpl Tortoise.Encodable do
def encode(%Package.Publish{identifier: nil, qos: 0} = t) do
[
Package.Meta.encode(%{t.__META__ | flags: encode_flags(t)}),
Package.variable_length_encode([
Package.length_encode(t.topic),
encode_payload(t)
])
]
end
def encode(%Package.Publish{identifier: identifier, qos: qos} = t)
when identifier in 0x0001..0xFFFF and qos in 1..2 do
[
Package.Meta.encode(%{t.__META__ | flags: encode_flags(t)}),
Package.variable_length_encode([
Package.length_encode(t.topic),
<<identifier::big-integer-size(16)>>,
encode_payload(t)
])
]
end
defp encode_flags(%{dup: dup, qos: qos, retain: retain}) do
<<flags::4>> = <<flag(dup)::1, qos::integer-size(2), flag(retain)::1>>
flags
end
defp encode_payload(%{payload: nil}), do: ""
defp encode_payload(%{payload: payload}), do: payload
defp flag(f) when f in [0, nil, false], do: 0
defp flag(_), do: 1
end
end
@@ -0,0 +1,28 @@
defmodule Tortoise.Package.Pubrec do
@moduledoc false
@opcode 5
alias Tortoise.Package
@opaque t :: %__MODULE__{
__META__: Package.Meta.t(),
identifier: Tortoise.package_identifier()
}
defstruct __META__: %Package.Meta{opcode: @opcode, flags: 0b000},
identifier: nil
@spec decode(<<_::32>>) :: t
def decode(<<@opcode::4, 0::4, 2, identifier::big-integer-size(16)>>)
when identifier in 0x0001..0xFFFF do
%__MODULE__{identifier: identifier}
end
# Protocols ----------------------------------------------------------
defimpl Tortoise.Encodable do
def encode(%Package.Pubrec{identifier: identifier} = t)
when identifier in 0x0001..0xFFFF do
[Package.Meta.encode(t.__META__), <<2, t.identifier::big-integer-size(16)>>]
end
end
end
@@ -0,0 +1,29 @@
defmodule Tortoise.Package.Pubrel do
@moduledoc false
@opcode 6
alias Tortoise.Package
@opaque t :: %__MODULE__{
__META__: Package.Meta.t(),
identifier: Tortoise.package_identifier()
}
@enforce_keys [:identifier]
defstruct __META__: %Package.Meta{opcode: @opcode, flags: 0b0010},
identifier: nil
@spec decode(<<_::32>>) :: t
def decode(<<@opcode::4, 2::4, 2, identifier::big-integer-size(16)>>)
when identifier in 0x0001..0xFFFF do
%__MODULE__{identifier: identifier}
end
# Protocols ----------------------------------------------------------
defimpl Tortoise.Encodable do
def encode(%Package.Pubrel{identifier: identifier} = t)
when identifier in 0x0001..0xFFFF do
[Package.Meta.encode(t.__META__), <<2, t.identifier::big-integer-size(16)>>]
end
end
end
@@ -0,0 +1,68 @@
defmodule Tortoise.Package.Suback do
@moduledoc false
@opcode 9
alias Tortoise.Package
@type qos :: 0 | 1 | 2
@type ack_result :: {:ok, qos} | {:error, :access_denied}
@opaque t :: %__MODULE__{
__META__: Package.Meta.t(),
identifier: Tortoise.package_identifier(),
acks: [ack_result]
}
@enforce_keys [:identifier]
defstruct __META__: %Package.Meta{opcode: @opcode, flags: 0},
identifier: nil,
acks: []
@spec decode(binary()) :: t
def decode(<<@opcode::4, 0::4, payload::binary>>) do
with payload <- drop_length(payload),
<<identifier::big-integer-size(16), acks::binary>> <- payload do
case return_codes_to_list(acks) do
[] ->
{:error, {:protocol_violation, :empty_subscription_ack}}
sub_acks ->
%__MODULE__{identifier: identifier, acks: sub_acks}
end
end
end
defp drop_length(payload) do
case payload do
<<0::1, _::7, r::binary>> -> r
<<1::1, _::7, 0::1, _::7, r::binary>> -> r
<<1::1, _::7, 1::1, _::7, 0::1, _::7, r::binary>> -> r
<<1::1, _::7, 1::1, _::7, 1::1, _::7, 0::1, _::7, r::binary>> -> r
end
end
defp return_codes_to_list(<<>>), do: []
defp return_codes_to_list(<<0x80::integer, acks::binary>>),
do: [{:error, :access_denied}] ++ return_codes_to_list(acks)
defp return_codes_to_list(<<ack::integer, acks::binary>>) when ack in 0x00..0x02,
do: [{:ok, ack}] ++ return_codes_to_list(acks)
# Protocols ----------------------------------------------------------
defimpl Tortoise.Encodable do
def encode(%Package.Suback{identifier: identifier} = t)
when identifier in 0x0001..0xFFFF do
[
Package.Meta.encode(t.__META__),
Package.variable_length_encode([
<<identifier::big-integer-size(16)>>,
Enum.map(t.acks, &encode_ack/1)
])
]
end
defp encode_ack({:ok, qos}) when qos in 0x00..0x02, do: qos
defp encode_ack({:error, _}), do: 0x80
end
end
@@ -0,0 +1,110 @@
defmodule Tortoise.Package.Subscribe do
@moduledoc false
@opcode 8
alias Tortoise.Package
@type qos :: 0 | 1 | 2
@type topic :: {binary(), qos}
@type topics :: [topic]
@opaque t :: %__MODULE__{
__META__: Package.Meta.t(),
identifier: Tortoise.package_identifier(),
topics: topics()
}
defstruct __META__: %Package.Meta{opcode: @opcode, flags: 0b0010},
identifier: nil,
topics: []
@spec decode(binary()) :: t
def decode(<<@opcode::4, 0b0010::4, length_prefixed_payload::binary>>) do
payload = drop_length(length_prefixed_payload)
<<identifier::big-integer-size(16), topics::binary>> = payload
topic_list = decode_topics(topics)
%__MODULE__{identifier: identifier, topics: topic_list}
end
defp drop_length(payload) do
case payload do
<<0::1, _::7, r::binary>> -> r
<<1::1, _::7, 0::1, _::7, r::binary>> -> r
<<1::1, _::7, 1::1, _::7, 0::1, _::7, r::binary>> -> r
<<1::1, _::7, 1::1, _::7, 1::1, _::7, 0::1, _::7, r::binary>> -> r
end
end
defp decode_topics(<<>>), do: []
defp decode_topics(<<length::big-integer-size(16), rest::binary>>) do
<<topic::binary-size(length), return_code::integer-size(8), rest::binary>> = rest
[{topic, return_code}] ++ decode_topics(rest)
end
# PROTOCOLS ==========================================================
defimpl Tortoise.Encodable do
def encode(
%Package.Subscribe{
identifier: identifier,
# a valid subscribe package has at least one topic/qos pair
topics: [{<<_topic_filter::binary>>, qos} | _]
} = t
)
when identifier in 0x0001..0xFFFF and qos in 0..2 do
[
Package.Meta.encode(t.__META__),
Package.variable_length_encode([
<<identifier::big-integer-size(16)>>,
encode_topics(t.topics)
])
]
end
defp encode_topics(topics) do
Enum.map(topics, fn {topic, qos} ->
[Package.length_encode(topic), <<0::6, qos::2>>]
end)
end
end
defimpl Enumerable do
def reduce(%Package.Subscribe{topics: topics}, acc, fun) do
Enumerable.List.reduce(topics, acc, fun)
end
def member?(%Package.Subscribe{topics: topics}, value) do
{:ok, Enum.member?(topics, value)}
end
def count(%Package.Subscribe{topics: topics}) do
{:ok, Enum.count(topics)}
end
def slice(_) do
# todo
{:error, __MODULE__}
end
end
defimpl Collectable do
def into(%Package.Subscribe{topics: topics} = source) do
{Enum.into(topics, %{}),
fn
acc, {:cont, {<<topic::binary>>, qos}} when qos in 0..2 ->
# if a topic filter repeat in the input we will pick the
# biggest one
Map.update(acc, topic, qos, &max(&1, qos))
acc, {:cont, <<topic::binary>>} ->
Map.put_new(acc, topic, 0)
acc, :done ->
%{source | topics: Map.to_list(acc)}
_, :halt ->
:ok
end}
end
end
end
@@ -0,0 +1,29 @@
defmodule Tortoise.Package.Unsuback do
@moduledoc false
@opcode 11
alias Tortoise.Package
@opaque t :: %__MODULE__{
__META__: Package.Meta.t(),
identifier: Tortoise.package_identifier()
}
@enforce_keys [:identifier]
defstruct __META__: %Package.Meta{opcode: @opcode, flags: 0b0000},
identifier: nil
@spec decode(<<_::32>>) :: t
def decode(<<@opcode::4, 0::4, 2, identifier::big-integer-size(16)>>)
when identifier in 0x0001..0xFFFF do
%__MODULE__{identifier: identifier}
end
# Protocols ----------------------------------------------------------
defimpl Tortoise.Encodable do
def encode(%Package.Unsuback{identifier: identifier} = t)
when identifier in 0x0001..0xFFFF do
[Package.Meta.encode(t.__META__), <<2, identifier::big-integer-size(16)>>]
end
end
end
@@ -0,0 +1,62 @@
defmodule Tortoise.Package.Unsubscribe do
@moduledoc false
@opcode 10
alias Tortoise.Package
@type topic :: binary()
@opaque t :: %__MODULE__{
__META__: Package.Meta.t(),
identifier: Tortoise.package_identifier(),
topics: [topic]
}
defstruct __META__: %Package.Meta{opcode: @opcode, flags: 2},
topics: [],
identifier: nil
@spec decode(binary()) :: t
def decode(<<@opcode::4, 0b0010::4, payload::binary>>) do
with payload <- drop_length(payload),
<<identifier::big-integer-size(16), topics::binary>> <- payload,
topic_list <- decode_topics(topics),
do: %__MODULE__{identifier: identifier, topics: topic_list}
end
defp drop_length(payload) do
case payload do
<<0::1, _::7, r::binary>> -> r
<<1::1, _::7, 0::1, _::7, r::binary>> -> r
<<1::1, _::7, 1::1, _::7, 0::1, _::7, r::binary>> -> r
<<1::1, _::7, 1::1, _::7, 1::1, _::7, 0::1, _::7, r::binary>> -> r
end
end
defp decode_topics(<<>>), do: []
defp decode_topics(<<length::big-integer-size(16), rest::binary>>) do
<<topic::binary-size(length), rest::binary>> = rest
[topic] ++ decode_topics(rest)
end
# Protocols ----------------------------------------------------------
defimpl Tortoise.Encodable do
def encode(
%Package.Unsubscribe{
identifier: identifier,
# a valid unsubscribe package has at least one topic filter
topics: [_topic_filter | _]
} = t
)
when identifier in 0x0001..0xFFFF do
[
Package.Meta.encode(t.__META__),
Package.variable_length_encode([
<<identifier::big-integer-size(16)>>,
Enum.map(t.topics, &Package.length_encode/1)
])
]
end
end
end
+176
View File
@@ -0,0 +1,176 @@
defmodule Tortoise.Pipe do
@moduledoc """
Experimental. This feature is under development.
The transmitter "pipe", for lack of a better word, is an opaque data
type that can be given to a process. It contains amongst other
things a socket.
A process can obtain a transmitter pipe by issuing a `pipe =
Tortoise.Pipe.new(client_id)` request, which will result in a pipe
in passive mode, meaning it will hold a socket it can publish
messages into, but might fail, in which case it will attempt to get
another socket from the transmitter. This all happens behind the
scenes, it is important though that the returned pipe is used in
future pipe requests, so publishing on a pipe should look like this:
pipe = Tortoise.Pipe.publish(pipe, "foo/bar", "bonjour !")
This is all experimental, and efforts to document this better will
be made when the design and implementation has stabilized.
"""
alias Tortoise.{Package, Pipe}
alias Tortoise.Connection.Inflight
@opaque t :: %__MODULE__{
client_id: binary(),
socket: port(),
transport: atom(),
active: boolean(),
failure: :crash | :drop,
timeout: non_neg_integer() | :infinity,
pending: [reference()]
}
@enforce_keys [:client_id]
defstruct([
:client_id,
socket: nil,
transport: Tortoise.Transport.Tcp,
active: false,
failure: :crash,
timeout: :infinity,
pending: []
])
@doc """
Create a new publisher pipe.
"""
def new(client_id, opts \\ []) do
active = Keyword.get(opts, :active, false)
timeout = Keyword.get(opts, :timeout, 5000)
opts = [timeout: timeout, active: active]
case Tortoise.Connection.connection(client_id, opts) do
{:ok, {transport, socket}} ->
%Pipe{client_id: client_id, transport: transport, socket: socket, active: active}
{:error, :unknown_connection} ->
{:error, :unknown_connection}
end
end
@doc """
Publish a message using a pipe.
"""
def publish(%Pipe{} = pipe, topic, payload \\ nil, opts \\ []) do
publish = %Package.Publish{
topic: topic,
payload: payload,
qos: Keyword.get(opts, :qos, 0),
retain: Keyword.get(opts, :retain, false)
}
with %Pipe{} = pipe <- do_publish(pipe, publish) do
pipe
else
{:error, :timeout} ->
# run pipe error spec
{:error, :timeout}
end
end
defp do_publish(%Pipe{} = pipe, %Package.Publish{qos: 0} = publish) do
encoded_publish = Package.encode(publish)
case pipe.transport.send(pipe.socket, encoded_publish) do
:ok ->
pipe
{:error, :closed} ->
case refresh(pipe) do
%Pipe{} = pipe ->
do_publish(pipe, publish)
{:error, :timeout} ->
{:error, :timeout}
end
end
end
defp do_publish(%Pipe{client_id: client_id} = pipe, %Package.Publish{qos: qos} = publish)
when qos in 1..2 do
case Inflight.track(client_id, {:outgoing, publish}) do
{:ok, ref} ->
updated_pending = [ref | pipe.pending]
%Pipe{pipe | pending: updated_pending}
end
end
defp refresh(%Pipe{active: true, client_id: client_id} = pipe) do
receive do
{{Tortoise, ^client_id}, :connection, {transport, socket}} ->
%Pipe{pipe | transport: transport, socket: socket}
after
pipe.timeout ->
{:error, :timeout}
end
end
defp refresh(%Pipe{active: false} = pipe) do
opts = [timeout: pipe.timeout, active: false]
case Tortoise.Connection.connection(pipe.client_id, opts) do
{:ok, {transport, socket}} ->
%Pipe{pipe | transport: transport, socket: socket}
{:error, :timeout} ->
{:error, :timeout}
end
end
@doc """
Await for acknowledge messages for the currently pending messages.
Note that this enters a selective receive loop, so the await needs
to happen before the process reaches its mailbox. It can be used in
situations where we want to send a couple of messages and continue
when the server has received them; This only works for messages with
a Quality of Service above 0.
"""
def await(pipe, timeout \\ 5000)
def await(%Pipe{pending: []} = pipe, _timeout) do
{:ok, pipe}
end
def await(%Pipe{client_id: client_id, pending: [ref | rest]} = pipe, timeout) do
receive do
{{Tortoise, ^client_id}, ^ref, :ok} ->
await(%Pipe{pipe | pending: rest})
after
timeout ->
{:error, :timeout}
end
end
# protocols
# defimpl Collectable do
# def into(pipe) do
# collector_fun = fn
# acc, {:cont, %Package.Publish{qos: 0} = elem} ->
# [Package.encode(elem) | acc]
# acc, :done ->
# Transmitter.publish(pipe, Enum.reverse(acc))
# acc
# _acc, :halt ->
# :ok
# end
# {[], collector_fun}
# end
# end
end
@@ -0,0 +1,39 @@
defmodule Tortoise.Registry do
@moduledoc false
@type via :: {:via, Registry, {__MODULE__, {module(), Tortoise.client_id()}}}
@type key :: atom() | tuple()
@spec via_name(module(), Tortoise.client_id()) :: via() | pid()
def via_name(_module, pid) when is_pid(pid), do: pid
def via_name(module, client_id) do
{:via, Registry, reg_name(module, client_id)}
end
@spec reg_name(module(), Tortoise.client_id()) :: {__MODULE__, {module(), Tortoise.client_id()}}
def reg_name(module, client_id) do
{__MODULE__, {module, client_id}}
end
@spec meta(key :: key()) :: {:ok, term()} | :error
def meta(key) do
Registry.meta(__MODULE__, key)
end
@spec put_meta(key :: key(), value :: term()) :: :ok
def put_meta(key, value) do
:ok = Registry.put_meta(__MODULE__, key, value)
end
@spec delete_meta(key :: term()) :: :ok | no_return
def delete_meta(key) do
try do
:ets.delete(__MODULE__, key)
:ok
catch
:error, :badarg ->
raise ArgumentError, "unknown registry: #{inspect(__MODULE__)}"
end
end
end
@@ -0,0 +1,65 @@
defmodule Tortoise.Supervisor do
@moduledoc """
A dynamic supervisor that can hold `Tortoise.Connection` processes
A `Tortoise.Supervisor`, registered under the name
`Tortoise.Supervisor`, is started as part of the `Tortoise.App`,
which will get started automatically when `Tortoise` is included as
a `Mix` dependency. Spawning a connection on the
`Tortoise.Supervisor` can be done as follows:
{:ok, pid} =
Tortoise.Supervisor.start_child(
client_id: T1000,
handler: {Tortoise.Handler.Logger, []},
server: {Tortoise.Transport.Tcp, host: 'localhost', port: 1883},
subscriptions: [{"foo/bar", 0}]
)
While this is an easy way to get started one should consider the
supervision strategy of the application using the
`Tortoise.Connection`s. Often one would like to close the connection
with the application using the connection, in which case one could
consider starting the `Tortoise.Connection` directly in the
supervisor of the application using the connection, or start a
`Tortoise.Supervisor` as part of ones application if it require more
than one connection.
See the *Connection Supervision* article in the project
documentation for more information on connection supervision.
"""
use DynamicSupervisor
@doc """
Start a dynamic supervisor that can hold connection processes.
The `:name` option can also be given in order to register a supervisor
name, the supported values are described in the "Name registration"
section in the `GenServer` module docs.
"""
def start_link(opts) do
DynamicSupervisor.start_link(__MODULE__, opts, name: opts[:name])
end
def child_spec(opts) do
opts = Keyword.put_new(opts, :name, __MODULE__)
DynamicSupervisor.child_spec(opts)
end
@doc """
Start a connection as a child of the `Tortoise.Supervisor`.
`supervisor` is the name of the supervisor the child should be
started on, and it defaults to `Tortoise.Supervisor`.
"""
def start_child(supervisor \\ __MODULE__, opts) do
spec = {Tortoise.Connection, opts}
DynamicSupervisor.start_child(supervisor, spec)
end
@impl true
def init(_opts) do
DynamicSupervisor.init(strategy: :one_for_one)
end
end
@@ -0,0 +1,71 @@
defmodule Tortoise.Transport do
@moduledoc """
Abstraction for working with network connections; this is done to
normalize the `:ssl` and `:gen_tcp` modules, so they get a similar
interface.
This work has been heavily inspired by the Ranch project by
NineNines.
"""
@opaque t :: %__MODULE__{
type: atom(),
host: binary(),
port: non_neg_integer(),
opts: [term()]
}
@enforce_keys [:type, :host, :port]
defstruct type: nil, host: nil, port: nil, opts: []
@doc """
Create a new Transport specification used by the Connection process
to log on to the MQTT server. This allow us to filter the options
passed to the connection type, and guide the user to connect to the
individual transport type.
"""
@spec new({atom(), [term()]}) :: t()
def new({transport, opts}) do
%Tortoise.Transport{type: ^transport} = transport.new(opts)
end
@type socket() :: any()
@type opts() :: any()
@type stats() :: any()
@callback new(opts()) :: Tortoise.Transport.t()
@callback listen(opts()) :: {:ok, socket()} | {:error, atom()}
@callback accept(socket(), timeout()) :: {:ok, socket()} | {:error, :closed | :timeout | atom()}
@callback accept_ack(socket(), timeout()) :: :ok
@callback connect(charlist(), :inet.port_number(), opts(), timeout()) ::
{:ok, socket()} | {:error, atom()}
@callback recv(socket(), non_neg_integer(), timeout()) ::
{:ok, any()} | {:error, :closed | :timeout | atom()}
@callback send(socket(), iodata()) :: :ok | {:error, atom()}
@callback setopts(socket(), opts()) :: :ok | {:error, atom()}
@callback getopts(socket(), [atom()]) :: {:ok, opts()} | {:error, atom()}
@callback getstat(socket()) :: {:ok, stats()} | {:error, atom()}
@callback getstat(socket(), [atom()]) :: {:ok, stats()} | {:error, atom()}
@callback controlling_process(socket(), pid()) :: :ok | {:error, :closed | :now_owner | atom()}
@callback peername(socket()) ::
{:ok, {:inet.ip_address(), :inet.port_number()}} | {:error, atom()}
@callback sockname(socket()) ::
{:ok, {:inet.ip_address(), :inet.port_number()}} | {:error, atom()}
@callback shutdown(socket(), :read | :write | :read_write) :: :ok | {:error, atom()}
@callback close(socket()) :: :ok
end
@@ -0,0 +1,130 @@
defmodule Tortoise.Transport.SSL do
@moduledoc false
@behaviour Tortoise.Transport
alias Tortoise.Transport
@default_opts [verify: :verify_peer]
@impl true
def new(opts) do
{host, opts} = Keyword.pop(opts, :host)
{port, opts} = Keyword.pop(opts, :port)
{list_opts, opts} = Keyword.pop(opts, :opts, [])
host = coerce_host(host)
opts = Keyword.merge(@default_opts, opts)
opts = [:binary, {:packet, :raw}, {:active, false} | opts] ++ list_opts
%Transport{type: __MODULE__, host: host, port: port, opts: opts}
end
defp coerce_host(host) when is_binary(host) do
String.to_charlist(host)
end
defp coerce_host(otherwise) do
otherwise
end
@impl true
def connect(host, port, opts, timeout) do
# [:binary, active: false, packet: :raw]
:ssl.connect(host, port, opts, timeout)
end
@impl true
def recv(socket, length, timeout) do
:ssl.recv(socket, length, timeout)
end
@impl true
def send(socket, data) do
:ssl.send(socket, data)
end
@impl true
def setopts(socket, opts) do
:ssl.setopts(socket, opts)
end
@impl true
def getopts(socket, opts) do
:ssl.getopts(socket, opts)
end
@impl true
def getstat(socket) do
:ssl.getstat(socket)
end
@impl true
def getstat(socket, opt_names) do
:ssl.getstat(socket, opt_names)
end
@impl true
def controlling_process(socket, pid) do
:ssl.controlling_process(socket, pid)
end
@impl true
def peername(socket) do
:ssl.peername(socket)
end
@impl true
def sockname(socket) do
:ssl.sockname(socket)
end
@impl true
def shutdown(socket, mode) when mode in [:read, :write, :read_write] do
:ssl.shutdown(socket, mode)
end
@impl true
def close(socket) do
:ssl.close(socket)
end
@impl true
def listen(opts) do
case Keyword.has_key?(opts, :cert) do
true ->
:ssl.listen(0, opts)
false ->
throw("Please specify the cert key for the SSL transport")
end
end
@impl true
def accept(listen_socket, timeout) do
:ssl.transport_accept(listen_socket, timeout)
end
@impl true
def accept_ack(client_socket, timeout) do
case :ssl.handshake(client_socket, timeout) do
{:ok, _} ->
:ok
{:ok, _, _} ->
:ok
# abnormal data sent to socket
{:error, {:tls_alert, _}} ->
:ok = close(client_socket)
exit(:normal)
# Socket most likely stopped responding, don't error out.
{:error, reason} when reason in [:timeout, :closed] ->
:ok = close(client_socket)
exit(:normal)
{:error, reason} ->
:ok = close(client_socket)
throw(reason)
end
end
end
@@ -0,0 +1,104 @@
defmodule Tortoise.Transport.Tcp do
@moduledoc false
@behaviour Tortoise.Transport
alias Tortoise.Transport
@impl true
def new(opts) do
{host, opts} = Keyword.pop(opts, :host)
{port, opts} = Keyword.pop(opts, :port, 1883)
{list_opts, opts} = Keyword.pop(opts, :opts, [])
host = coerce_host(host)
opts = [:binary, {:packet, :raw}, {:active, false} | opts] ++ list_opts
%Transport{type: __MODULE__, host: host, port: port, opts: opts}
end
defp coerce_host(host) when is_binary(host) do
String.to_charlist(host)
end
defp coerce_host(otherwise) do
otherwise
end
@impl true
def connect(host, port, opts, timeout) do
# forced_opts = [:binary, active: false, packet: :raw]
# opts = Keyword.merge(opts, forced_opts)
:gen_tcp.connect(host, port, opts, timeout)
end
@impl true
def recv(socket, length, timeout) do
:gen_tcp.recv(socket, length, timeout)
end
@impl true
def send(socket, data) do
:gen_tcp.send(socket, data)
end
@impl true
def setopts(socket, opts) do
:inet.setopts(socket, opts)
end
@impl true
def getopts(socket, opts) do
:inet.getopts(socket, opts)
end
@impl true
def getstat(socket) do
:inet.getstat(socket)
end
@impl true
def getstat(socket, opt_names) do
:inet.getstat(socket, opt_names)
end
@impl true
def controlling_process(socket, pid) do
:gen_tcp.controlling_process(socket, pid)
end
@impl true
def peername(socket) do
:inet.peername(socket)
end
@impl true
def sockname(socket) do
:inet.sockname(socket)
end
@impl true
def shutdown(socket, mode) when mode in [:read, :write, :read_write] do
:gen_tcp.shutdown(socket, mode)
end
@impl true
def close(socket) do
:gen_tcp.close(socket)
end
@impl true
def listen(opts) do
# forced_opts = [:binary, active: false, packet: :raw, reuseaddr: true]
# opts = Keyword.merge(opts, forced_opts)
:gen_tcp.listen(0, opts)
end
@impl true
def accept(listen_socket, timeout) do
:gen_tcp.accept(listen_socket, timeout)
end
@impl true
def accept_ack(_socket, _timeout) do
:ok
end
end
+86
View File
@@ -0,0 +1,86 @@
defmodule Tortoise.MixProject do
use Mix.Project
@version "0.9.9"
def project do
[
app: :tortoise,
version: @version,
elixir: "~> 1.6",
start_permanent: Mix.env() == :prod,
description: description(),
package: package(),
deps: deps(),
dialyzer: dialyzer(),
docs: docs(),
test_coverage: [tool: ExCoveralls],
preferred_cli_env: [
coveralls: :test,
"coveralls.detail": :test,
"coveralls.html": :test,
"coveralls.json": :test,
"coveralls.post": :test,
docs: :docs
]
]
end
defp description() do
"""
A MQTT client for Elixir.
"""
end
# Run "mix help compile.app" to learn about applications.
def application do
[
extra_applications: [:logger, :ssl],
mod: {Tortoise.App, []}
]
end
# Run "mix help deps" to learn about dependencies.
defp deps do
[
{:gen_state_machine, "~> 2.0 or ~> 3.0"},
{:dialyxir, "~> 1.0.0-rc.3", only: [:dev], runtime: false},
{:eqc_ex, "~> 1.4", only: :test},
{:excoveralls, "~> 0.10", only: :test},
{:ex_doc, "~> 0.19", only: :dev, runtime: false},
{:ct_helper, github: "ninenines/ct_helper", only: :test}
]
end
defp package() do
[
maintainers: ["Martin Gausby"],
licenses: ["Apache 2.0"],
files: ["lib", "mix.exs", "README*", "CHANGELOG*", "LICENSE*"],
links: %{"GitHub" => "https://github.com/gausby/tortoise"}
]
end
defp docs() do
[
name: "Tortoise",
source_ref: "v#{@version}",
main: "introduction",
canonical: "http://hexdocs.pm/tortoise",
source_url: "https://github.com/gausby/tortoise",
extras: [
"docs/introduction.md",
"docs/connecting_to_a_mqtt_broker.md",
"docs/connection_supervision.md",
"docs/publishing_messages.md"
]
]
end
defp dialyzer() do
[
ignore_warnings: "dialyzer.ignore",
flags: [:error_handling, :race_conditions, :underspecs]
]
end
end