DEV Community: Behnam Nasehi

PlaySho: Virtual Cinema for Shared Movie Nights ( Android Kotlin )

Behnam Nasehi — Thu, 29 Feb 2024 09:31:28 +0000

Turn off those movie nights into immersive cinematic experiences with PlaySho! 🎬 You can watch the same video with your friends from anywhere. Create a virtual cinema, share a streaming link and enjoy synchronized playback to have unforgettable experience of watching a movie.

Introduction

PlaySho is an app built on Kotlin that transforms the movie night experience using advanced technologies, like Socket.IO, end to end encryption, hybrid encryption and ExoPlayer. Its main goal is to create a cinema setting where users can watch a video together in sync no matter where they are.

By utilizing Socket.IO PlaySho allows real time communication between users to synchronize video playback seamlessly and foster a sense of unity among viewers. Additionally the use of end to end encryption ensures user data and communications are private and secure building trust in the platform.

Hybrid encryption methods add a layer of security by combining asymmetric encryption techniques to protect sensitive information shared within the app.

With ExoPlayer at its core PlaySho offers a captivating multimedia experience with video playback on devices and platforms. Its cross platform compatibility allows users to enjoy movie nights without limitations, on devices, tablets or desktop computers.
Moreover PlaySho provides movie theaters enabling users to customize their settings according to their liking and establish a setting that enhances the enjoyment of watching movies. Through bringing people across distances and enabling shared entertainment PlaySho seeks to transform how users interact and bond with their friends and family during movie gatherings.

Key Features:

🎥 Virtual Cinema:

Set up your own private cinema room and have a shared movie night with friends.

🍿 Single Video Playback:

For that cinema-like feel, you can stream one video synchronously.

🔗 Stream Link Sharing:

Create a special stream link and invite your friends to join in your virtual theatre.

💬 Real-Time Interaction:

Chatting and sharing reactions among friends gives an opportunity to watch movies socially.

📱 Cross-Platform Accessibility:

Use PlaySho on different gadgets, be it smartphone, tablet or desktop computer to have an adjustable and joyful movie night.

🌐 Stay Connected:

Have common movie minutes with friends and family who are in different geographical locations.

🛠️ How It Works:

Create Cinema Room:
- Set up a virtual cinema room that can be altered to suit the right mood for a perfect movie night.
Choose Your Movie:
- Choose one video that will be enjoyed by all people together.
Generate Stream Link:
- Share a unique streaming link with your buddies.
Watch Together:
- Experience the magic of synchronized playback as you watch a single video together in real-time.

🌟 Why PlaySho:

Virtual Movie Nights:
- Make shared film nights feel like you are actually at the theatre or cinema hall
Synchronized Playback:
- Have perfect timing and synchronization that is characteristic of watching movies from cinemas
No Limits:
- Friends might be watching from anywhere making distances irrelevant

Ways to Contribute

1. Reporting Bugs

If you encounter any bugs or issues while using PlaySho, please report them on our GitHub Issues page . Be sure to provide details about the case, including steps to deliver and surroundings.

2. Feature Requests

Have an idea for a new feature or improvement? Feel free to share it with us on our GitHub Issues page. We welcome your suggestions and will consider them for future updates.

3. Code Contributions

For developers interested in contributing code to PlaySho, follow these steps:

- Fork the Repository

Click on the "Fork" button at the top of the PlaySho repository to create your fork.

- Clone Your Fork

Fork the repository to your GitHub account by clicking on the "Fork" button on the top right of the repository page.

Next, clone your fork to your local machine using the following command:

git clone https://github.com/behnamnasehi/playsho.git

cd playsho

- Create a New Branch

Create a new branch for your feature or bug fix:

git checkout -b feature/your-feature-name

- Make Changes

Make your code changes and ensure they adhere to our coding standards.

- Test Locally

Test your changes locally to ensure they work as expected.

- Commit and Push

git add .

git commit -m "Add your commit message here"

git push origin feature/your-feature-name

- Create a Pull Request

Visit the Pull Requests page in your repository and click on "New Pull Request." Provide a clear description of your changes and submit the pull request.

4. Documentation

Improving documentation is another valuable way to contribute. If you identify areas where the documents can be enlarged or have written guidelines, feel free to submit a pull request.

5. Design Contributions

Designers are also welcome to contribute to the project. Whether it’s improving the user interface, creating graphics, or improving the overall aesthetic of the design, your contributions are appreciated.

License

By contributing to PlaySho, you agree that your contributions will be licensed under the Apache-2.0 license license.

As the lead Android developer in the back of PlaySho,Behnam Nasehi, I'm captivated with crafting revolutionary solutions that beautify user reviews and push the bounds of what is viable within the realm of mobile applications. With a strong basis in Kotlin, Socket.IO, and superior encryption techniques, I attempt to create immersive experiences that bring human beings together, whether or not they're sharing film nights or taking part on initiatives. I'm committed to staying at the leading edge of Android improvement developments and technology, constantly seeking new possibilities to innovate and deliver extraordinary effects. Connect with Behnam Nasehi to explore how we are able to elevate your subsequent cell project collectively!

Here's everything you need to know about MQTT Broker (simple explanation)

Behnam Nasehi — Mon, 27 Feb 2023 10:23:34 +0000

The MQTT broker helps devices communicate with each other like a messenger. It's like a digital postman, except it goes "beep" instead of "ding dong"! Imagine that you have a group of friends who want to share messages with each other. They can't just shout at each other because they might be far away or not in the same room. So, they need a messenger to deliver their messages.

The MQTT broker is like that messenger. It helps different devices like smartphones, sensors, and other machines talk to each other over the internet. Each device sends a message to the broker, and the broker delivers that message to the device or devices that need to receive it.

So, just like a messenger helps friends communicate with each other, the MQTT broker helps devices communicate with each other over the internet.

Websocket vs MQTT: what's the difference
Here's how MQTT brokers work
The Publish Method
Subscribe Method
real-world scenario example
QoS (Quality of Service)
Security
MQTT broker in Nest.js
Contact Me

Websocket vs MQTT: what's the difference

Websockets and MQTT are both protocols used for communication over the internet, but they have some differences in how they work and what they are best suited for.

Web sockets are designed for real-time bidirectional communication between a client and a server. They establish a long-term connection between the two, allowing for low-latency communication and efficient data transfer. Websockets are often used in web applications for tasks like live chat or real-time data visualization.

MQTT, on the other hand, is a lightweight publish-subscribe messaging protocol that is optimized for machine-to-machine communication in IoT (Internet of Things) applications. It is designed to be efficient and reliable, even in low-bandwidth or unreliable network conditions. MQTT uses a broker to mediate communication between multiple clients, making it well-suited for applications where many devices need to send and receive data with low overhead and minimal latency.

In summary, while both Websockets and MQTT can be used for real-time communication, Websockets are typically used for client-server communication in web applications, while MQTT is used for machine-to-machine communication in IoT applications.

Here's how MQTT brokers work

MQTT broker is a central communication hub that facilitates communication between devices using the MQTT protocol. It receives messages published by devices and delivers them to devices that have subscribed to receive them. The MQTT broker works based on the publish-subscribe pattern, which means that devices publish messages to a specific topic, and other devices subscribe to that topic to receive those messages.

Here is how the MQTT broker works in more detail:

Devices publish messages to the MQTT broker by connecting to it and specifying a topic to publish to. Each message has a payload (the data being sent) and a topic (a string that identifies the message).
The MQTT broker receives the message and stores it in a buffer until it can be delivered to subscribers. The broker checks the topic of the message and compares it to the list of subscribed topics.
Subscribers can connect to the MQTT broker and subscribe to specific topics by specifying the topic they are interested in. Once subscribed, the MQTT broker delivers any messages published on that topic to the subscriber.
When the MQTT broker receives a message that matches a subscribed topic, it sends the message to all subscribers that have subscribed to that topic.

The Publish Method

Imagine you have a magic toy telephone that can talk to your friend's toy telephone. When you want to tell your friend something, you pick up your magic phone and say your message to it. Then you press a button to send the message to your friend's magic phone.

In MQTT, the publish method works the same way as your magic toy telephone. When you want to send a message to someone else's device (like a smartwatch or a thermostat), you use the publish method. You start by picking a topic for your message, which is like picking a phone number for your friend's toy phone. Then, you enter your message into the publish method and press a button to send it to the other device.

The other device, which is subscribed to the same topic as your message, will receive your message and can take action based on what you said.

Subscribe Method

Imagine you have a special mailbox that only opens when someone sends you a letter with a specific code written on it. When you want to receive messages from someone, you give them the code and they send you a letter with the code on it. Then, when the letter arrives in the mailbox you have set up, it opens and you can read the message inside.

In MQTT, the subscribe method works the same way as your special mailbox. When you want to receive messages from another device (like a sensor or a smart light), you use the subscribe method. You start by choosing a topic that you're interested in, which is like choosing a special code for your mailbox. Then you use the subscribe method to tell the MQTT broker that you want to receive any messages that are published on that topic.

When a message is published on a topic that you subscribe to, the MQTT broker sends it to your device so you can take action based on the message. It's like opening your special mailbox and reading the message that someone sent you!

To illustrate how an MQTT broker can be used in a real-world scenario, here's an example:

Let's say you have a smart home system with multiple sensors and devices, such as a thermostat, a smart lock, and smart lightbulbs. These devices can communicate with each other and with a central hub using the MQTT protocol.

When the temperature in your home drops below a certain threshold, the temperature sensor publishes a message to the MQTT broker with the topic "temperature" and the temperature reading as the payload. The thermostat subscribes to the "temperature" topic and receives the message from the MQTT broker. It then adjusts the heating system to increase the temperature in your home.

If you are away from home and want to let someone in, you can use the smartphone app to send a message to the MQTT broker with the topic "smart-lock" and the command to unlock the door as the payload. The smart lock subscribes to the "smart-lock" topic and receives the message from the MQTT broker. It then unlocks the door, allowing the person to enter.

In the evening, when you want to switch on the lights, you can use the smartphone app to send a message to the MQTT broker with the topic "smart-lights" and the command to turn on the lights as the payload. The smart lightbulbs subscribe to the "smart-lights" topic and receive the message from the MQTT broker. They then turn on, illuminating your home.

In this example, the temperature sensor, smart lock, and smart lightbulbs are all publishers that send messages to the MQTT broker on different topics. The thermostat, smart lock, and smart lightbulbs are subscribers that receive messages from the MQTT broker based on the topics they have subscribed to. The smartphone app acts as a publisher that sends commands to the MQTT broker with different topics, which are then received by the smart lock and smart lightbulbs. The MQTT broker acts as a mediator, delivering messages between the devices and the smartphone app.

What is QoS (Quality of Service) in the MQTT protocol?

QoS (Quality of Service) is like asking your friend to give you something, but with different levels of confirmation.

Level 0 :is like asking your friend to give you something, but not waiting for them to confirm that they heard you or that they will do it. You just assume they will do it and move on.

Level 1 :is like asking your friend to give you something. They confirm that they heard you and will do it, but they might forget or something might happen and they might not actually do it. So you might ask again later to make sure.

Level 2 :is like asking your friend to give you something. They confirm that they heard you and will do it. They keep track of it to make sure it gets done. This way, you can be sure that you will get what you asked for.

Similarly, in MQTT, QoS is used to determine the level of confirmation required for message delivery between devices. A higher QoS level means increased assurance of successful message delivery, but also increased overhead and potential network congestion.

Security of the MQTT protocol

The MQTT protocol has several security mechanisms in place to protect the confidentiality, integrity, and availability of data exchanged between devices. Here are some of the key security features of MQTT:

Encryption: MQTT supports encryption of data using the SSL/TLS protocol. This provides secure communication between devices and the MQTT broker, preventing eavesdropping and unauthorized access to sensitive information.
Authentication: MQTT allows for authentication of devices and users using various mechanisms, including usernames and passwords, X.509 certificates, and JSON Web Tokens (JWTs). This ensures that only authorized devices can publish and subscribe to topics on the MQTT broker.
Access Control: MQTT brokers can implement access control mechanisms to restrict access to topics based on the identity of the device or user, or based on the content of the message being published. This helps prevent unauthorized access and ensures that sensitive data is only visible to authorized parties.
Message Filtering: MQTT brokers can implement message filtering to prevent the delivery of messages to unauthorized subscribers. This helps prevent unauthorized access to sensitive data and reduces network bandwidth usage.
Quality of Service (QoS): MQTT provides different levels of QoS for message delivery, ranging from at-most-once delivery (QoS 0) to at-least-once delivery (QoS 1) to exactly-once delivery (QoS 2). The higher levels of QoS provide increased reliability and consistency in message delivery, ensuring that messages are not lost or duplicated.

Overall, the security features of MQTT make it a suitable protocol for use in IoT applications where security and privacy are critical concerns. However, it is imperative to properly configure and implement security measures to ensure the effectiveness of these features.

What is the best way to build an MQTT broker in Nest.js?

pigeon-mqtt-nest is an NPM package that simplifies the process of integrating an MQTT broker into Nest JS applications.

Using the MQTT.js library, pigeon-mqtt-nest provides a wrapper that enables seamless integration with Nest JS applications. The package provides both MQTT client and server functionality, allowing developers to customize the behavior of the MQTT broker according to their specific requirements.

One of the key benefits of using pigeon-mqtt-nest is that it enables real-time communication between devices and services, which is particularly helpful for IoT applications where data needs to be exchanged in real-time. With pigeon-mqtt-nest, Nest JS developers can easily add MQTT functionality to their applications, without having to worry about the low-level details of MQTT protocol implementation.

Overall, pigeon-mqtt-nest is a valuable tool for Nest JS developers looking to integrate MQTT functionality into their applications. The package is easy to use, customizable, and provides a range of features that make it an excellent choice for building real-time IoT applications.

NPM link : https://www.npmjs.com/package/pigeon-mqtt-nest

Contact Me

Thank you to everyone who took the time to read this article! I hope you find the information useful and informative. The world of IoT and MQTT is constantly evolving, and I am grateful for the opportunity to share my knowledge with you. Now, go ahead and show off your new found knowledge by making your own IoT projects and, who knows, maybe you'll be the one to change the world!

If you have questions or feedback, please feel free to reach out. I would be happy to help in any way I can. Thank you again for your time and attention!

My linkedin : Check out my profile here. And while you’re at it, why not give me a follow and make my day?

How to build your own MQTT broker in Nest.js

Behnam Nasehi — Wed, 12 Oct 2022 06:22:21 +0000

What is MQTT ?

MQTT (Message Queuing Telemetry Transport) is an open source, lightweight messaging protocol, optimized for low latency. This protocol provides a callable and cost-efficient way to connect devices using a publish/subscribe model. A communication system built on MQTT consists of the publishing server, a broker and one or more clients. It is designed for constrained devices and low-bandwidth, high-latency or unreliable networks.

Pigeon MQTT Nestjs

Pigeon MQTT Nestjs is broker that can run on any stream server

Installation

Install From NPM :

$ npm i pigeon-mqtt-nest

Usage

Pigeon will register as a global module. You can
import with configuration

@Module({
  imports: [

    PigeonModule.forRoot({
      port: 1884,
      id: "binarybeast",
      concurrency:100,
      queueLimit:42,
      maxClientsIdLength:23,
      connectTimeout:30000,
      heartbeatInterval:60000,
     })

  ],
  controllers: [AppController],
  providers: [AppService],
})
export class AppModule {
}

options <object>
- mq <MQEmitter> middleware used to deliver messages to subscribed clients. In a cluster environment it is used also to share messages between brokers instances. Default: mqemitter
- concurrency <number> maximum number of concurrent messages delivered by mq. Default: 100
- persistence <Persistence> middleware that stores QoS > 0, retained, will packets and subscriptions. Default: aedes-persistence (in memory)
- queueLimit <number> maximum number of queued messages before client session is established. If number of queued items exceeds, connectionError throws an error Client queue limit reached. Default: 42
- maxClientsIdLength option to override MQTT 3.1.0 clients Id length limit. Default: 23
- heartbeatInterval <number> an interval in millisconds at which server beats its health signal in $SYS/<aedes.id>/heartbeat topic. Default: 60000
- id <string> aedes broker unique identifier. Default: uuidv4()
- connectTimeout <number> maximum waiting time in milliseconds waiting for a [CONNECT][CONNECT] packet. Default: 30000

Handlers

Handler	Emitted When
preConnect	Invoked when server receives a valid `CONNECT` packet.
authenticate	Invoked after `preConnect`.
authorizePublish	publish LWT to all online clients,incoming client publish
authorizeSubscribe	restore subscriptions in non-clean session.,incoming client `SUBSCRIBE`
published	same as `Event: publish`, but provides a backpressure functionality.

Handler: preConnect

client: <Client>
packet: <object> [CONNECT][CONNECT]
callback: <Function> (error, successful) => void
- error <Error> | null
- successful <boolean>

Invoked when server receives a valid [CONNECT][CONNECT] packet. The packet can be modified.

client object is in default state. If invoked callback with no errors and successful be true, server will
continue to establish a session.

Any error will be raised in connectionError event.

Some Use Cases:

Rate Limit / Throttle by client.conn.remoteAddress
Check aedes.connectedClient to limit maximum connections
IP blacklisting

@Injectable()
export class TestService {

  @onPreConnect()
  onPreConnect(@Client() client, @Packet() packet, @Function() done) {
    console.log("Function: @onPreConnect()");
    return done(null, true);
  }

}

Handler: authenticate

client: <Client>
credential: <string>
callback: <Function> (error, successful) => void
- error <Error> | null
- successful <boolean>

Invoked after preConnect.

Server parses the [CONNECT][CONNECT] packet, initializes client object which set client.id to match the one
in [CONNECT][CONNECT] packet and extract username and password as parameters for user-defined authentication flow.

If invoked callback with no errors and successful be true, server authenticates client and continues to setup
the client session.

If authenticated, server acknowledges a [CONNACK][CONNACK] with returnCode=0, otherwise returnCode=5. Users could
define the value between 2 and 5 by defining a returnCode property in error object.

@Injectable()
export class TestService {

  @onAuthenticate()
  onAuthenticate(@Client() client, @Credential() credential, @Function() done) {
    console.log("Function: @onAuthenticate()");
    return done(null, true);
  }

}

@Injectable()
export class TestService {

  @onAuthenticate()
  onAuthenticate(@Client() client, @Credential() credential, @Function() done) {
    console.log("Function: @onAuthenticate()");
    var error = new Error('Auth error')
    error.returnCode = 4
    return done(error, false);
  }

}

Please refer to Connect Return Code
to see their meanings.

Handler: authorizePublish

client: <Client> | null
packet: <object> [PUBLISH][PUBLISH]
callback: <Function> (error) => void
- error <Error> | null

Invoked when

publish LWT to all online clients
incoming client publish

client is null when aedes publishes obsolete LWT without connected clients

If invoked callback with no errors, server authorizes the packet otherwise emits clientError with error. If
an error occurs the client connection will be closed, but no error is returned to the client (MQTT-3.3.5-2)

@Injectable()
export class TestService {

  @onAuthorizePublish()
  onAuthorizePublish(@Client() client, @Packet() packet, @Function() done) {
    console.log("Function: @onAuthorizePublish()");
    if (packet.topic === 'aaaa') {
      return done(new Error('wrong topic'))
    }
    if (packet.topic === 'bbb') {
      packet.payload = Buffer.from('overwrite packet payload')
    }
    return done(null);
  }

}

By default authorizePublish throws error in case a client publish to topics with $SYS/ prefix to prevent possible
DoS (see #597). If you write your own implementation
of authorizePublish we suggest you to add a check for this. Default implementation:

@Injectable()
export class TestService {

  @onAuthorizePublish()
  onAuthorizePublish(@Client() client, @Packet() packet, @Function() done) {
    if (packet.topic.startsWith($SYS_PREFIX)) {
      return done(new Error($SYS_PREFIX + ' topic is reserved'))
    }
    return done(null);
  }

}

Handler: authorizeSubscribe

client: <Client>
subscription: <object>
callback: <Function> (error) => void
- error <Error> | null
- subscription: <object> | null

Invoked when

restore subscriptions in non-clean session.
incoming client [SUBSCRIBE][SUBSCRIBE]

subscription is a dictionary object like { topic: hello, qos: 0 }.

If invoked callback with no errors, server authorizes the packet otherwise emits clientError with error.

In general user should not touch the subscription and pass to callback, but server gives an option to change the
subscription on-the-fly.

@Injectable()
export class TestService {

  @onAuthorizeSubscribe()
  onAuthorizeSubscribe(@Client() client, @Subscription() subscription, @Function() done) {
    console.log("Function: @onAuthorizeSubscribe()");
    if (subscription.topic === 'aaaa') {
      return done(new Error('wrong topic'))
    }
    if (subscription.topic === 'bbb') {
      // overwrites subscription
      subscription.topic = 'foo'
      subscription.qos = 1
    }
    return done(null, subscription);
  }

}

To negate a subscription, set the subscription to null. Aedes ignores the negated subscription and the qos
in SubAck is set to 128 based
on MQTT 3.11 spec:

@Injectable()
export class TestService {

  @onAuthorizeSubscribe()
  onAuthorizeSubscribe(@Client() client, @Subscription() subscription, @Function() done) {
    done(null, subscription.topic === 'aaaa' ? null : sub)
  }

}

Handler: published

packet: <aedes-packet> & [PUBLISH][PUBLISH]
client: <Client>
callback: <Function>

same as Event: publish, but provides a backpressure functionality. TLDR; If you are doing operations
on packets that MUST require finishing operations on a packet before handling the next one use this otherwise,
especially for long-running operations, you should use Event: publish instead.

@Injectable()
export class TestService {

  @onPublish()
  OnPublish(@Topic() topic, @Packet() packet, @Payload() payload, @Client() client) {
    console.log("Function: @OnPublish()");
  }

}

Events

Name	Emitted When
Client	`client` registers itself to `server`
Client Ready	`client` has received all its offline messages and be initialized.
Client Disconnect	`client` disconnects.
Client Error	an error occurs.
Connection Error	an error occurs.
Keep Alive Timeout	timeout happes in the `client` keepalive.
Publish	`servers` delivers the `packet` to subscribed `client`.
Ack	`packet` successfully delivered to the `client`.
Subscribe	`client` successfully subscribe the `subscriptions` in server.
Unsubscribe	`client` successfully unsubscribe the `subscriptions` in server.
Connack Sent	`server` sends an acknowledge to `client`.
Closed	`server` is closed.

Event: client

client <Client>

Emitted when the client registers itself to server. The client is not ready yet.
Its connecting state equals to true.

Server publishes a SYS topic $SYS/<aedes.id>/new/clients to inform it registers the client into its registration
pool. client.id is the payload.

@Injectable()
export class TestService {

  constructor(@Inject(PigeonService) private readonly aedesService: PigeonService) {
  }

  @onClient()
  OnNewClient(@Client() client) {
    console.log("Function: @onClient()");
  }

}

Event: clientReady

client <Client>

Emitted when the client has received all its offline messages and be initialized.
The client connected state equals to true and is ready for processing incoming
messages.

@Injectable()
export class TestService {

  constructor(@Inject(PigeonService) private readonly aedesService: PigeonService) {
  }

  @onClientReady()
  async onClientReady(@Client() client) {
    console.log("Function: @onClientReady()");
  }

}

Event: clientDisconnect

client <Client>

Emitted when a client disconnects.

Server publishes a SYS topic $SYS/<aedes.id>/disconnect/clients to inform it deregisters the client. client.id is
the payload.

@Injectable()
export class TestService {

  constructor(@Inject(PigeonService) private readonly aedesService: PigeonService) {
  }

  @onClientDisconnect()
  OnClientDisconnect(@Client() client) {
    console.log("Function: @OnClientDisconnect()");
  }

}

Event: clientError

client <Client>
error <Error>

Emitted when an error occurs.

@Injectable()
export class TestService {

  constructor(@Inject(PigeonService) private readonly aedesService: PigeonService) {
  }

  @onClientError()
  OnClientError(@Client() client, @Error() error) {
    console.log("Function: @onClientError()");
  }

}

Event: connectionError

client <Client>
error <Error>

Emitted when an error occurs. Unlike clientError it raises only when client is uninitialized.

@Injectable()
export class TestService {

  constructor(@Inject(PigeonService) private readonly aedesService: PigeonService) {
  }

  @onConnectionError()
  OnConnectionError(@Client() client, @Error() error) {
    console.log("Function: @OnConnectionError()");
  }

}

Event: keepaliveTimeout

client <Client>

Emitted when timeout happes in the client keepalive.

@Injectable()
export class TestService {

  constructor(@Inject(PigeonService) private readonly aedesService: PigeonService) {
  }

  @onKeepLiveTimeout()
  onKeepLiveTimeout(@Client() client) {
    console.log("Function: @onKeepLiveTimeout()");
  }

}

Event: publish

packet <aedes-packet> & [PUBLISH][PUBLISH]
client <Client> | null

Emitted when servers delivers the packet to subscribed client. If there are no clients subscribed to the packet
topic, server still publish the packet and emit the event. client is null when packet is an internal message
like aedes heartbeat message and LWT.

Note! packet belongs aedes-packet type. Some properties belongs to aedes internal, any changes on them will break aedes internal flow.

@Injectable()
export class TestService {

  constructor(@Inject(PigeonService) private readonly aedesService: PigeonService) {
  }

  @onPublish()
  OnPublish(@Topic() topic, @Packet() packet, @Payload() payload, @Client() client) {
    console.log("Function: @OnPublish()");
  }

}

Event: ack

packet <object> [PUBLISH][PUBLISH] for QoS 1, [PUBREL][PUBREL] for QoS 2
client <Client>

Emitted an QoS 1 or 2 acknowledgement when the packet successfully delivered to the client.

@Injectable()
export class TestService {

  constructor(@Inject(PigeonService) private readonly aedesService: PigeonService) {
  }


  @onAck()
  onAck(@Client() client, @Packet() packet) {
    console.log("Function: @onAck()");
  }

}

Event: subscribe

subscriptions <object>
client <Client>

Emitted when client successfully subscribe the subscriptions in server.

subscriptions is an array of { topic: topic, qos: qos }. The array excludes duplicated topics and includes negated
subscriptions where qos equals to 128. See more
on authorizeSubscribe

Server publishes a SYS topic $SYS/<aedes.id>/new/subscribers to inform a client successfully subscribed to one or more
topics. The payload is a JSON that has clientId and subs props, subs equals to subscriptions array.

@Injectable()
export class TestService {

  constructor(@Inject(PigeonService) private readonly aedesService: PigeonService) {
  }

  @onSubscribe()
  OnSubscribe(@Subscription() subscription, @Client() client) {
    console.log("Function: @OnSubscribe()");
  }

}

Event: unsubscribe

unsubscriptions Array<string>
client <Client>

Emitted when client successfully unsubscribe the subscriptions in server.

unsubscriptions are an array of unsubscribed topics.

Server publishes a SYS topic $SYS/<aedes.id>/new/unsubscribers to inform a client successfully unsubscribed to one or
more topics. The payload is a JSON that has clientId and subs props, subs equals to unsubscriptions array.

@Injectable()
export class TestService {

  constructor(@Inject(PigeonService) private readonly aedesService: PigeonService) {
  }

  @onUnsubscribe()
  OnUnsubscribe(@Subscription() subscription, @Client() client) {
    console.log("Function: @OnUnsubscribe()");
  }

}

Event: connackSent

packet <object> [CONNACK][CONNACK]
client <Client>

Emitted when server sends an acknowledge to client. Please refer to the MQTT specification for the explanation of
returnCode object property in CONNACK.

@Injectable()
export class TestService {

  constructor(@Inject(PigeonService) private readonly aedesService: PigeonService) {
  }

  @onConnackSent()
  onConnackSent(@Client() client, @Packet() packet) {
    console.log("Function: @onConnackSent()");
  }

}

Event: closed

Emitted when server is closed.

@Injectable()
export class TestService {

  constructor(@Inject(PigeonService) private readonly aedesService: PigeonService) {
  }

  @onClosed()
  onClosed(@Client() client, @Packet() packet) {
    console.log("Function: @onClosed()");
  }

}

Methods

Method	Emitted When
Publish	Directly deliver `packet` on behalf of server to subscribed clients.
Close	Close aedes server and disconnects all clients.

Method: Publish

packet <object> [PUBLISH][PUBLISH]

Directly deliver packet on behalf of server to subscribed clients.
Bypass authorizePublish.

callback will be invoked with error arugments after finish.

@Injectable()
export class TestService {

  //inject Pigeon Service
  constructor(@Inject(PigeonService) private readonly pigeonService: PigeonService) {
  }

  @onPublish()
  async OnPublish(@Topic() topic, @Packet() packet, @Payload() payload, @Client() client) {

    //use this method to publish
    await this.pigeonService.publish({
      topic: "test2", qos: 0, cmd: "publish", payload: "", dup: false, retain: false
    });

  }

}

Method: Close

callback: <Function>

Close aedes server and disconnects all clients.

callback will be invoked when server is closed.

@Injectable()
export class TestService {

  //inject Pigeon Service
  constructor(@Inject(PigeonService) private readonly pigeonService: PigeonService) {
  }

  @onPublish()
  async OnPublish(@Topic() topic, @Packet() packet, @Payload() payload, @Client() client) {

    //use this method to publish
    await this.pigeonService.close();

  }

}

Method: Get Broker Instance

get broker instance

@Injectable()
export class TestService {

  //inject Pigeon Service
  constructor(@Inject(PigeonService) private readonly pigeonService: PigeonService) {
  }

  @onPublish()
  async OnPublish(@Topic() topic, @Packet() packet, @Payload() payload, @Client() client) {

    await this.pigeonService.getBrokerInstance();

  }

}

Packets

This section describes the format of all packets

Packet	-
Connect	---
Connack	---
Subscribe	---
Suback	---
Unsubscribe	---
Unsuback	---
Publish	---
Puback	---
Pubrec	---
Pubrel	---
Pubcomp	---
Pingreq	---
Pingresp	---
Disconnect	---

Packet: Connect

{
  cmd: 'connect',
  protocolId: 'MQTT', // Or 'MQIsdp' in MQTT 3.1 and 5.0
  protocolVersion: 4, // Or 3 in MQTT 3.1, or 5 in MQTT 5.0
  clean: true, // Can also be false
  clientId: 'my-device',
  keepalive: 0, // Seconds which can be any positive number, with 0 as the default setting
  username: 'matteo',
  password: Buffer.from('collina'), // Passwords are buffers
  will: {
    topic: 'mydevice/status',
    payload: Buffer.from('dead'), // Payloads are buffers
    properties: { // MQTT 5.0
      willDelayInterval: 1234,
      payloadFormatIndicator: false,
      messageExpiryInterval: 4321,
      contentType: 'test',
      responseTopic: 'topic',
      correlationData: Buffer.from([1, 2, 3, 4]),
      userProperties: {
        'test': 'test'
      }
    }
  },
  properties: { // MQTT 5.0 properties
      sessionExpiryInterval: 1234,
      receiveMaximum: 432,
      maximumPacketSize: 100,
      topicAliasMaximum: 456,
      requestResponseInformation: true,
      requestProblemInformation: true,
      userProperties: {
        'test': 'test'
      },
      authenticationMethod: 'test',
      authenticationData: Buffer.from([1, 2, 3, 4])
  }
}

If password or will.payload are passed as strings, they will automatically be converted into a Buffer.

Packet: Connack

{
  cmd: 'connack',
  returnCode: 0, // Or whatever else you see fit MQTT < 5.0
  sessionPresent: false, // Can also be true.
  reasonCode: 0, // reason code MQTT 5.0
  properties: { // MQTT 5.0 properties
      sessionExpiryInterval: 1234,
      receiveMaximum: 432,
      maximumQoS: 2,
      retainAvailable: true,
      maximumPacketSize: 100,
      assignedClientIdentifier: 'test',
      topicAliasMaximum: 456,
      reasonString: 'test',
      userProperties: {
        'test': 'test'
      },
      wildcardSubscriptionAvailable: true,
      subscriptionIdentifiersAvailable: true,
      sharedSubscriptionAvailable: false,
      serverKeepAlive: 1234,
      responseInformation: 'test',
      serverReference: 'test',
      authenticationMethod: 'test',
      authenticationData: Buffer.from([1, 2, 3, 4])
  }
}

Packet: Subscribe

{
  cmd: 'subscribe',
  messageId: 42,
  properties: { // MQTT 5.0 properties
    subscriptionIdentifier: 145,
    userProperties: {
      test: 'test'
    }
  }
  subscriptions: [{
    topic: 'test',
    qos: 0,
    nl: false, // no Local MQTT 5.0 flag
    rap: true, // Retain as Published MQTT 5.0 flag
    rh: 1 // Retain Handling MQTT 5.0
  }]
}

Packet: Suback

{
  cmd: 'suback',
  messageId: 42,
  properties: { // MQTT 5.0 properties
    reasonString: 'test',
    userProperties: {
      'test': 'test'
    }
  }
  granted: [0, 1, 2, 128]
}

Packet: Unsubscribe

{
  cmd: 'unsubscribe',
  messageId: 42,
  properties: { // MQTT 5.0 properties
    userProperties: {
      'test': 'test'
    }
  }
  unsubscriptions: [
    'test',
    'a/topic'
  ]
}

Packet: Unsuback

{
  cmd: 'unsuback',
  messageId: 42,
  properties: { // MQTT 5.0 properties
    reasonString: 'test',
    userProperties: {
      'test': 'test'
    }
  }
}

Packet: Publish

{
  cmd: 'publish',
          messageId: 42,
          qos: 2,
          dup: false,
          topic: 'test',
          payload: Buffer.from('test'),
          retain: false,
          properties: { // optional properties MQTT 5.0
    payloadFormatIndicator: true,
            messageExpiryInterval: 4321,
            topicAlias: 100,
            responseTopic: 'topic',
            correlationData: Buffer.from([1, 2, 3, 4]),
            userProperties: {
      'test': 'test'
    },
    subscriptionIdentifier: 120, // can be an Array in message from broker, if message included in few another subscriptions
            contentType: 'test'
  }
}

Packet: Puback

{
  cmd: 'puback',
          messageId: 42,
          reasonCode: 16, // only for MQTT 5.0
          properties: { // MQTT 5.0 properties
    reasonString: 'test',
            userProperties: {
      'test': 'test'
    }
  }
}

Packet: Pubrec

{
  cmd: 'pubrec',
          messageId: 42,
          reasonCode: 16, // only for MQTT 5.0
          properties: { // properties MQTT 5.0
    reasonString: 'test',
            userProperties: {
      'test': 'test'
    }
  }
}

Packet: Pubrel

{
  cmd: 'pubrel',
          messageId: 42,
          reasonCode: 16, // only for MQTT 5.0
          properties: { // properties MQTT 5.0
    reasonString: 'test',
            userProperties: {
      'test': 'test'
    }
  }
}

Packet: Pubcomp

{
  cmd: 'pubcomp',
          messageId: 42,
          reasonCode: 16, // only for MQTT 5.0
          properties: { // properties MQTT 5.0
    reasonString: 'test',
            userProperties: {
      'test': 'test'
    }
  }
}

Packet: Pingreq

{
  cmd: 'pingreq'
}

Packet: Pingresp

{
  cmd: 'pingresp'
}

Packet: Disconnect

{
  cmd: 'disconnect',
          reasonCode: 0, // MQTT 5.0 code
          properties: { // properties MQTT 5.0
    sessionExpiryInterval: 145,
            reasonString: 'test',
            userProperties: {
      'test': 'test'
    },
    serverReference: 'test'
  }
}

Middleware Plugins

Persistence

aedes-persistence: In-memory implementation of an Aedes persistence
aedes-persistence-mongodb: MongoDB persistence for Aedes
aedes-persistence-redis: Redis persistence for Aedes
aedes-persistence-level: LevelDB persistence for Aedes
aedes-persistence-nedb: NeDB persistence for Aedes

MQEmitter

mqemitter: An opinionated memory Message Queue with an emitter-style API
mqemitter-redis: Redis-powered mqemitter
mqemitter-mongodb: Mongodb based mqemitter
mqemitter-child-process: Share the same mqemitter between a hierarchy of child processes
mqemitter-cs: Expose a MQEmitter via a simple client/server protocol
mqemitter-p2p: A P2P implementation of MQEmitter, based on HyperEmitter and a Merkle DAG
mqemitter-aerospike: Aerospike mqemitter

Dependencies

Stay in touch

Author Twitter - @binarybeast
Author Linkedin - behnamnasehi