Ingest data from Solace

You can ingest data from Solace’s PubSub+ Platform, a powerful event-driven streaming solution designed for real-time enterprises.

It facilitates the design, deployment, integration, and management of event-driven architectures (EDAs) across hybrid, multi-cloud, and IoT environments. It enables seamless data exchange across legacy systems, SaaS applications, messaging services, databases, and AI agents, connecting them to a real-time event-driven layer.

​

Set up Solace

To set up Solace PubSub+ event broker, you can either choose the free Software version using Docker or Solace PubSub+ Cloud.

​

Scenario

Consider this scenario: automating the process of notifying passengers that “online check-in is open” exactly 48 hours before their flight departure. Airlines need to handle continuous streams of flight and passenger data to send timely “Check-in Open” alerts to passengers who have opted in. The process begins 72 hours before departure, as flight and passenger data enter the system. Then, at 48 hours before departure, a notification is triggered for eligible passengers.

The solution involves two key steps:

1. Event Stream Processing: Continuous streams of flight and passenger data are received from the Departure Control System (DCS) via Solace. Each flight is tracked by a unique identifier, and each passenger by a unique Passenger Reference Number (PRN), enabling real-time processing in RisingWave.
2. Notification Logic: Notifications are sent only to passengers who have opted in.

Below is the sample data of flight and passenger details. Solace topic: passenger_full_details

{
        "passenger_ref_number": "PRN026",
        "flight_id": "LH6456",
        "flight_number": "6456",
        "carrier_code": "LH",
        "flight_date": "2024-10-17",
        "origin": "LHR",
        "departure_time": "2024-10-17T04:40:00Z",
        "contact_info": "john.garcia@gmail.com",
        "opted_in": true
    }

​

Ingest data from Solace into RisingWave

Create a RisingWave cluster in RisingWave Cloud using the free plan. See the documentation of RisingWave Cloud for instructions.

​

Step 1: Create source table

This query creates a table named combined_passenger_flight_data to store detailed passenger and flight information. The data is sourced from the Solace topic passenger_full_details, connected through the Solace broker, with the Quality of Service (QoS) level set to at least once and formatted as plain JSON.

CREATE TABLE combined_passenger_flight_data (
    flight_id VARCHAR,
    flight_number VARCHAR,
    carrier_code VARCHAR,
    flight_date DATE,
    origin VARCHAR,
    passenger_ref_number VARCHAR,
    departure_time TIMESTAMPTZ,
    opted_in BOOLEAN,
    contact_info VARCHAR
)
WITH (
    connector = 'mqtt',
    topic = 'passenger_full_details',
    url = 'ssl://xxxxxxxxxx:8883',
    username='solace-cloud-client',
    password='xxxxxxxxxxxx', 
    qos = 'at_least_once'
) FORMAT PLAIN ENCODE JSON;

​

Step 2: Use materialized view to filter

This query creates a materialized view named checkin_open_notification that selects flight and passenger information for those who opted in and have flights departing within 48 to 72 hours from the current time.

CREATE MATERIALIZED VIEW checkin_open_notification AS
SELECT flight_id, passenger_ref_number, flight_number, carrier_code, departure_time, contact_info
FROM combined_passenger_flight_data
WHERE opted_in = TRUE
  AND departure_time <= NOW() - INTERVAL '48 hours'
  AND departure_time > NOW() - INTERVAL '72 hours';

​

Step 3: Query the materialized view

The materialized view can be queried to retrieve the latest data from the source:

SELECT * FROM checkin_open_notification LIMIT 5;

The table chart lists passengers who opted in for notifications and have flights departing soon, showing flight_id, passenger_ref_number, flight_number, carrier_code, departure_time, and contact_info. It highlights passengers with flights departing before 48 from now, indicating that check-in is open.

​

Step 4: Create a sink to send notifications

This query creates a sink named checkin_notifications_sink, which streams data from the checkin_open_notification view to the Solace topic checkin_open_notification. The connection to the Solace server is established with at-least-once Quality of Service (QoS), and the data is formatted as plain JSON. The online check-in notification system then retrieves this information from the Solace topic to send notifications to the passengers.

CREATE SINK checkin_notifications_sink
FROM checkin_open_notification
WITH (
    connector = 'mqtt',
    topic = 'checkin_open_notification',
    url = 'ssl://xxxxxxxxxx:8883',
    username='solace-cloud-client',
    password='xxxxxxxxxxxx', 
    qos = 'at_least_once'
) FORMAT PLAIN ENCODE JSON;

We have successfully created a source table to read data from the Solace with an MQTT source connector, built a materialized view (MV) for querying and performing real-time analytics on the data, and set up a sink to send processed data to a Solace topic using the MQTT sink connector for downstream systems to utilize.