DEV Community: Enrico Bellanti

Notification Brodacast System with Laravel-Websocket and Rxjs

Enrico Bellanti — Mon, 20 Jun 2022 13:17:07 +0000

Why Laravel and Rxjs?

This is my first post and i would like to explain how i resolve my issue.
Obviously you can find many tutorial to implement a websocket but sometimes you can get into my same situation when you are working with Laravel at back-end and Angular at front-end more specifically with Rxjs.
I tried to find some tutorial online which suggest to use laravel-websockets and laravel-echo which is a very common pattern but if you are using Rxjs is not the best solution.

Installing Laravel Websockets
Require the Laravel Websockets package. It works as a replacement for external services like Pusher. Many settings will refer to Pusher today but be reminded that we are not using it. We want our own solution.

composer require beyondcode/laravel-websockets

We also need a package by Pusher.

composer require pusher/pusher-php-server

Next, adapt your .env file. We want the BROADCAST_DRIVER to be pusher.

BROADCAST_DRIVER=pusher

And we need to set the Pusher credentials.
(Note: Again I want to mention that we do not use the Pusher service. Our websockets server just has the same API.)

PUSHER_APP_ID=12345
PUSHER_APP_KEY=12345
PUSHER_APP_SECRET=12345
PUSHER_APP_CLUSTER=mt1

The Laravel Websockets package comes with a migration file for storing statistics and a config file we need to adapt. Let's publish them.

php artisan vendor:publish --provider="BeyondCode\LaravelWebSockets\WebSocketsServiceProvider" --tag="migrations"

This will create a new migration file that we can run. Make sure you have set up a database for this project and defined the DB credentials in the .env file. Afterward, we can run the migration.

php artisan migrate

And here, we publish the config file of Laravel Websockets.

php artisan websockets:serve

To test that it is running, we can check the debugging dashboard under the endpoint /laravel-websockets. You can click connect to see if the dashboard can connect to the WebSockets server.

After clicking connect, you should see that the dashboard is subscribed to some debugging channels like private-websockets-dashboard-api-message. This will tell you that the server is set up correctly.

Broadcast Notifications From Our Laravel Application
We can use notifications to send data to our WebSockets server. So let's create a new one.

php artisan make:notification RealTimeNotification

Here is what we need to change:

use the ShouldBroadcast interface
add a message property which we will pass through the constructor
use the broadcast channel in the via method
add a toBroadcast method to define the message

<?php

namespace App\Notifications;

use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Broadcasting\ShouldBroadcast;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Notifications\Messages\MailMessage;
use Illuminate\Notifications\Notification;
use Illuminate\Notifications\Messages\BroadcastMessage;

class RealTimeNotification extends Notification implements ShouldBroadcast
{
    use ShouldQueue;

    public string $message;

    public function __construct(string $message)
    {
        $this->message = $message;
    }

    public function via($notifiable): array
    {
        return ['broadcast'];
    }

    public function toBroadcast($notifiable): BroadcastMessage
    {
        return new BroadcastMessage([
            'message' => "$this->message (User $notifiable->id)",
            'event' => class_basename($this),
        ]);
    }

}

Before we can try sending this event, please adapt your broadcasting.php config file to use the following options:

'options' => [
    'cluster' => env('PUSHER_APP_CLUSTER'),
    'encrypted' => false,
    'host' => '127.0.0.1',
    'port' => 6001,
    'scheme' => 'http'
],

With these options, we make sure that when we broadcast something from our Laravel application, it gets sent to our WebSockets server.

Let's have a test if everything is working properly.
Let's trigger the nofification and check in the websockets dashbord if you have sucess.

$user = User::first();

$user->notify(new App\Notifications\RealTimeNotification('Hello World'));

Time to connect back-end to fornt-end

Let's have a look at code to see how to connect front-end with Rxjs with laravel-websockets

import { ChangeDetectionStrategy, Component, OnDestroy } from '@angular/core';
import { Store } from '@ngrx/store';
import { filter, map, Subject, takeUntil } from 'rxjs';
import { IAuthTokenWs } from './shared/interfaces/webSocket.interface';
import { WebsocketService } from './shared/services/webSocket/websocket.service';
import { selectUserId } from './store/user/user-feature.selectors';

@Component({
  selector: 'hh-root',
  template: `
    <bn-loading-spinner></bn-loading-spinner>
    <router-outlet></router-outlet>
  `,
  styleUrls: ['./app.component.scss'],
  changeDetection: ChangeDetectionStrategy.OnPush,
})
export class AppComponent implements OnDestroy {
  destroy$$ = new Subject<void>();

  constructor(private websocketService: WebsocketService, private store: Store) {
    this.store
      .select(selectUserId)
      .pipe(
        takeUntil(this.destroy$$),
        filter((e) => !!e),
        map((user_id) => {
          let socket$ = new WebSocket('ws://localhost:6001/app/12345');

          socket$.onmessage = (msg) => {
            let obj = JSON.parse(msg.data);

            if (obj.event === 'pusher:connection_established') {
              let socket_id = JSON.parse(obj.data).socket_id;
              this.websocketService
                .authWebsocket(user_id!!, socket_id)
                .pipe(takeUntil(this.destroy$$))
                .subscribe((e: IAuthTokenWs) => {
                  let tmp = JSON.stringify({
                    event: 'pusher:subscribe',
                    data: {
                      auth: e.auth,
                      channel: `private-App.Models.User.${user_id}`,
                    },
                  });
                  socket$.send(tmp);
                });
            } else {
              this.websocketService.notifications.next([
                ...this.websocketService.notifications.value,
                JSON.parse(obj?.data)?.message,
              ]);
              console.log(JSON.parse(obj?.data)?.message);
            }
          };
        }),
      )
      .subscribe();
  }

  ngOnDestroy(): void {
    this.destroy$$.next();
  }
}

You probably noticed that we did not define a channel name with our notification as we did in our event. This is because there is a default pattern for the channel name of a notification notifiable-class.key. In our case, this would be App.Models.User.1. And when you take a look at the Web dashboard, you find a message triggered by our notification to the channel Channel: private-App.Models.User.1.

Broadcast::channel('App.Models.User.{id}', function ($user, $id) {
    return (int) $user->id === (int) $id;
});

In our case we are using Ngrx aswell so we will get user_id with
.select(selectUserId), but obiusly you can get anywhere you have stored it.
Also in this application we use an interceptor which attach token to api call to be authorize by laravel guard at backend when we need to call the following api to retrive token auth to listen on private channel via websocket.

  authWebsocket(userID: string, socket_id: string) {
    const data = {
      socket_id,
      channel_name: `private-App.Models.User.${userID}`,
    };

    return this.http.post<IAuthTokenWs>(`${this.apiBaseUrl}/broadcasting/auth`, data);
  }

After implemented it when triggering notification at back-end you should recive a console.log in your application front-end.
Now your BE and FE are connected.

Considering you can receive notification in broadcast just if you are connected to the web application in real time broadcasting, probably when you are offline you won't get them back after you reconnect, there are several methods to implement this and i would like to make soon a new short tutorial "how to store notification unseen and retrives in specific moment" (for example immediatelly after login).

Note: consider to implement Laravel Queue System with notifications to avoid that your faild ones never delivered to you receivers. I just wrote a an additional guide for it.notification-queue-tutorial

Laravel Notification System with Queue

Enrico Bellanti — Mon, 20 Jun 2022 13:10:00 +0000

Why Laravel Queue?

As i explain in previus posts i really suggest when you are using notification to implement also a queue system to avoid that failed notifications never get delivered.
Reading the queue notification for first time seems to be something complicated to be implemented but it isn't so hard.
I'll try to help you with this tutorial to make it easier as possible.

Let's start
In order to use Queue you need to migrate table jobs with laravel command

php artisan queue:table

php artisan migrate

and then set the .ENV file with database

QUEUE_CONNECTION=database

Let's now create a notification example and see how to implement interface and trait

<?php

namespace App\Notifications;

use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Broadcasting\ShouldBroadcast;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Notifications\Notification;
use Illuminate\Notifications\Messages\BroadcastMessage;

class RealTimeNotification extends Notification implements ShouldBroadcast
{
    use Queueable;

    public string $message;

    public function __construct(string $message)
    {
        $this->message = $message;
    }

    public function via($notifiable): array
    {
        return ['broadcast'];
    }

    public function toBroadcast($notifiable): BroadcastMessage
    {
        return new BroadcastMessage([
            'message' => "$this->message"
        ]);
    }

    /**
     * Get the array representation of the notification.
     *
     * @param  mixed  $notifiable
     * @return array
     */
    public function toArray($notifiable)
    {
        return [
            'message' => "$this->message",
        ];
    }
}

with the method toArray we can specify how the notification must be formatted to be saved in the queue. Also this method is used to save the notification in notifications database table (I have already explained this feature in the post Notification System with websocket )

One you add in notification class the interfrace "ShouldBroadcast" and the trait "Queueable" you can automatically send to job table all the notifications and in case they fail you can continue to manage without loosing forever, i will show you forward in this tutorial how to do that.

Artisan queue:work
From now you will have two table in database jobs and failed_jobs ones you trigger a notification it will be delivered firstly on jobs table and managed in queue if one of these fails it will be moved to failed_jobs table avoid to be get lost.

To manage all these processes you need to start the artisan command

php artisan queue:work

after this command you can specify several configurations to be set at most but for this i would suggest to have a look at laravel docs
laravel-queue-doc

The artisan queue:work get one by one your notifications and send them according to your needs.
you have here an example of configs

php artisan queue:work --sleep=3 --tries=3 --backoff=3 --max-time=3600 --max-jobs=5 --stop-when-empty****

As you can see i just set --tries 3, for example, in this way you can make the system tries 3 times to send notification before it get failed and move to failed_jobs table.

Sometimes jobs can fail for a variety of resons, such as an exceeded worker timeout, in these case you can manage these in a different time, you can do that just running the command

php artisan queue:retry

in this way you just pull back your jobs to the jobs table and make queue:work reprocess for a second time and give it a second chance.

!attention: all these processes are run manually in this tutorial but should be managed by your server, i would suggest to install a supervisor and just add the configuration that laravel provide in its documentation
supervisor configuration

installing supervisor is not so difficult and after that you will have a supervisor.conf file where to add some settings like this

[program:laravel-worker]
process_name=%(program_name)s_%(process_num)02d
command=php /var/www/html/artisan queue:work --sleep=3 --tries=3 --backoff=3 --max-time=3600 --max-jobs=5 --stop-when-empty
autostart=true
autorestart=true
stopasgroup=true
killasgroup=true
startretries=3
user=root
numprocs=1
redirect_stderr=true
stdout_logfile=/var/www/html/storage/logs/worker.log
stderr_logfile=/var/www/html/storage/logs/worker-error.log
stopwaitsecs=3600

Hope this tutorial would be helpful but if i missed something or you have some ideas to refactor please contact me at twitter-personal-link.