Live Communications

Pegasus supports real-time live communications using WebSockets, allowing developers to subscribe to and receive events from Pegasus devices dynamically. This removes the need for polling (periodic HTTP requests), reducing bandwidth usage and improving efficiency.

When to Use Live Communications

• Use live communications when you need real-time updates for a single vehicle, asset, or a small group of entities.
• Avoid subscribing to a large number of vehicles or making high-frequency subscription requests, as this may trigger rate limits and affect the proper functioning of the application.
• If you need to receive large-scale, real-time data, consider using Forwarders, which allow receiving data via POST requests instead of WebSockets in real-time.

Key Benefits of Live Communications:

• Real-time data updates with minimal latency
• Reduced server load compared to polling
• Automatic reconnection handling
• Support for compressed payloads to minimize bandwidth
• Flexible subscription model for different event types

Requirements:

• Socket.IO Client v2.5.0
• WebSocket as a transport to successfully connect (polling is disabled)
• Valid authentication token from Pegasus Gateway

Connection Limits:

• Maximum of 250 entities per subscription when using cache
• Rate limiting applies to prevent system abuse
• IP ban enforcement for repeated invalid connection attempts

Getting Started

Installation

Add Socket.IO Client (v2.5.0) to your project

<script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/2.5.0/socket.io.js"></script>

or via npm

npm install socket.io-client@2.5.0

and import it in your application

const io = require('socket.io-client');

Basic Setup

You need to connect to the live_url found in your Pegasus Site by going to the site_url /api endpoint.
For example: https://cloud.pegasusgateway.com/api/

{
    "domain": "pegasus1.pegasusgateway.com",
    "name": "7.5.4",
    "pegasus_id": 1,
    "live_url": "https://aws-live-3.pegasusgateway.com",
    "tag": "7.5.4",
    "date": "Jan 23 09:31:08 2025"
}

// Create a new WebSocket connection to your Pegasus live server
const socket = io('https://aws-live-3.pegasusgateway.com/socket', {
    transports: ['websocket']  // Important: polling is disabled
});

// Set up basic event handlers
socket.on('connect', () => {
    console.log('Connected to Pegasus Live server');
});

socket.on('disconnect', () => {
    console.log('Disconnected from Pegasus Live server');
});

Authentication

Before receiving any events, you must authenticate with your Pegasus Gateway credentials, or a valid API token:

// Set up credentials
const credentials = {
    pegasus: "https://your-pegasus-gateway.com",
    auth: "your-auth-token"  // Get this from /api/login endpoint
};

socket.emit("authenticate", credentials);

socket.on('_authenticated', (data) => {
    console.log('Successfully authenticated', data);
    // data contains available resources (vehicles, assets)
});

socket.on('_error', (message) => {
    console.error('Authentication error:', message);
});

Subscribe to Events

After authentication, subscribe to events for specific vehicles:

// Subscribe to events for a single vehicle
socket.on('events', (envelope) => {
    console.log('Received event:', envelope);
    // Handle event data here
});

// Subscribe to vehicle events
const subscription = {
    namespace: "vehicle-events",
    objects: 123,  // Vehicle ID
    compact: true  // Enable compressed payloads
};

socket.emit("listen", subscription);

Core Concepts

Namespaces

Each event from Pegasus falls under a specific namespace, which determines its category. Understanding these namespaces helps you filter and process relevant events effectively

Available Namespaces

1. vehicle-events: Events generated by vehicles, such as tracking updates, status changes, or input/output changes.
2. jobs: Events related to job execution (e.g., job started, completed, or failed).
3. triggers: Events associated with trigger actions, such as alerts or automated responses.

Event Structure

The main event payload contains:

• namespace: Identifies the category of the event.
• object: The ID of the entity (e.g., vehicle) that generated the event.
• payload: The detailed data of the event, including device status, event type, and other related information.

Event Flow

1. Device Generation: Vehicle devices generate events
2. Server Processing: Pegasus processes and enriches the data
3. Subscription Filtering: Events are filtered based on your subscriptions
4. Delivery: Matching events are sent to your WebSocket connection

// Example of subscribing to multiple namespaces
const vehicleSubscription = {
    namespace: "vehicle-events",
    objects: 123,
    compact: true
};

const triggerSubscription = {
    namespace: "triggers",
    objects: "all"  // Listen to all triggers you have access to
};

// Subscribe to both
socket.emit("listen", vehicleSubscription);
socket.emit("listen", triggerSubscription);

Message Types

Pegasus Live Communications uses events to deliver real-time updates to your application.

Incoming Messages (server → client)

System Messages

• _authenticated: Authentication success, includes output of resources handler
• _error: Error notifications (authentication not successful, missing credentials or permissions, etc.)
• _update: General status updates
• resources: Available resource list (entity ids, triggers, etc.)
• events: Primary event data channel

// Handling different message types
socket.on('_authenticated', (data) => {
    console.log('Available resources:', data);
});

socket.on('_update', (message) => {
    console.log('System update:', message);
});

socket.on('events', (envelope) => {
    const { namespace, object, payload } = envelope;
    console.log(`${namespace} event for object ${object}:`, payload);
});

Outgoing Messages (client → server)

Control Messages

• authenticate: Send authentication credentials
• listen: Subscribe to events
• stop: Unsubscribe from events
• resources: Request available resources

// Example of all outgoing message types
function controlExample() {
    // Authentication
    socket.emit('authenticate', {
        pegasus: "https://your-gateway.com",
        auth: "your-token"
    });

    // Request resources
    socket.emit('resources');

    // Subscribe to events
    socket.emit('listen', {
        namespace: "vehicle-events",
        objects: 123
    });

    // Unsubscribe from events
    socket.emit('stop', {
        namespace: "vehicle-events",
        objects: 123
    });
}

Event Envelope Structure

Every event comes wrapped in an envelope containing metadata and payload:

// Example event envelope structure
{
    namespace: "vehicle-events",       // Event category
    object: 123,                      // Entity ID (e.g., vehicle ID)
    payload: {                        // Actual event data
        device: {                     // Device state
            connection: {
                online: true
                // ... other connection details
            }
            // ... other device details
        },
        event: {                      // Event details
            // ... event specific data
        },
        updates: ["device", "event"]  // What was updated
    }
}

// Handling event envelopes
socket.on('events', (envelope) => {
    // Check what type of update this is
    if (envelope.updates.includes('event')) {
        // Handle new event data
        handleNewEvent(envelope.payload.event);
    }
    if (envelope.updates.includes('device')) {
        // Handle device state changes
        handleDeviceUpdate(envelope.payload.device);
    }
});

Event Handling

Device State Updates

The device object contains current state information:

function handleDeviceState(payload) {
    const { device } = payload;
    
    // Check connection state
    if (device.connection) {
        const isOnline = device.connection.online;
        console.log(`Device is ${isOnline ? 'online' : 'offline'}`);
    }
    
    // Check GPS validity
    if (device.gpsknit) {
        const hasValidGPS = !device.gpsknit.no_location_count;
        console.log(`GPS is ${hasValidGPS ? 'valid' : 'invalid'}`);
    }
    
    // Check I/O states
    if (device.ios_state) {
        console.log('Input states:', device.ios_state);
    }
}

Event Updates

The event object contains the latest telemetry data:

function handleEventData(payload) {
    const { event } = payload;
    
    // Basic event processing
    const eventTime = new Date(event.event_time * 1000);
    console.log('Event occurred at:', eventTime);
    
    // Location data (if available)
    if (event.lat && event.lon) {
        console.log('Location and speed:', {
            latitude: event.lat,  
            longitude: event.lon,
            kph: event.mph * 1.60934
        });
    }
}

Implementation Requirements

Pegasus Live Communications provides several advanced features to improve efficiency, reduce bandwidth usage, and handle large-scale data processing. It is necessary to apply these so that your application is not rated for improper use.

1. Compact Event Merging – Optimizing event handling by reducing redundant data.
2. Paginating Subscriptions – Handling large numbers of vehicles efficiently.
3. Using Cache Callbacks – Managing real-time event streams efficiently.

1. Compact Event Merging

By default, event payloads contain a full dataset, which can be large and inefficient. Compact mode solves this by only sending changed fields.

To enable compact mode, set "compact": true in your subscription request:

const subscriptionEnvelope = {
    "namespace": "vehicle-events",
    "objects": [617, 197], // Replace with vehicle IDs
    "compact": true
};

socket.emit("listen", subscriptionEnvelope);

Handling Compact Events

When compact mode is enabled, you need to merge compact updates with the previously received full event to reconstruct the complete data.

// Utils and libraries
import _mergeWith from 'lodash.mergewith'
const ws_merger = (objValue, srcValue, key, object, source, stack) => {
  if (Array.isArray(objValue) && Array.isArray(srcValue)) return srcValue;
  return undefined;
};
const __cache = {}
// receive cached payload, update with compressed update, and return full event
const digest_ws_event = (update) => {
	const cached = __cache[update.primary_id] || {}
	__cache[update.primary_id] = _mergeWith(cached, update, ws_merger)
  const live_update = __cache[update.primary_id]
	let { updates, event, _event_restore } = update
	if (updates.includes('event')) {
		const { event: cevent } = live_update
		live_update.event = event
		_event_restore ??= []
		_event_restore.map((k) => {
			live_update.event[k] = cevent[k] || cached.device.latest.data?.[k]?.value
		})
	return live_update
	}
}
// event callback
const process_event = (update) => {
	const payload = digest_ws_event(update.payload)
	// application code
}

// cache callback
const cache_callback = (events) => events.map(process_event)

// attach function to events handler
socket.on('events', (event) => {
	process_event(event)
})

2. Paginating Subscriptions

Pegasus enforces a limit of 250 entities per subscription when using the cache_callback function. If you need to subscribe to a large number of vehicles, you must paginate the requests.

import _chunk from 'lodash.chunk'

const $sleep = (ms) => new Promise((resolve, reject) => setTimeout(resolve, ms))

socket.on('resources', async function(resources){
  // resources.vehicles and resources.assets represent all ids that you can subscribe to
  for (const resource of ['vehicles', 'assets']) {
    for (const chunk of _chunk(resources[resource], 100)) {
      const envelope = {
        "namespace":"vehicle-events",
        "objects": chunk,
        "compact": true
      }
      socket.emit('listen', envelope, cache_callback)
      await $sleep(50) // recommended
    }
  }
})

3. Using Cache Callbacks

For high-frequency event subscriptions, cache callbacks help store, merge, and process updates efficiently.

How Cache Callbacks Work

• Initial event → Stores a full event payload in a cache.
• Subsequent compact updates → Merges into the cached data.
• Processed data → Sent to your application.

const processEvent = (update) => {
    const fullEvent = mergeCompactEvent(update.object, update.payload);
    console.log("Processed event:", fullEvent);
};

const cacheCallback = (events) => {
    events.forEach(processEvent);
};

// Subscribe with cache handling
const envelope = {
    "namespace": "vehicle-events",
    "objects": [617, 197],
    "compact": true
};

socket.emit("listen", envelope, cacheCallback);

Practical Example

In practice this is what the information above looks like with respect to the full event payload received once you subscribe to vehicle-events.

Full payload

Confirmation you are subscribed

Further events are received partially and can be merged with the full payload to obtain the entire updated package.

Stopping a Subscription

To unsubscribe from specific events, send a stop request with a subscription envelope similar to the one used for subscribing.

const unsubscribeEnvelope = {
    "namespace": "vehicle-events",
    "objects": [617, 197] // Replace with vehicle IDs to stop listening
};

// Stop listening to events for these vehicles
socket.emit("stop", unsubscribeEnvelope);

⚠️ Important:

• This only stops the current session's subscriptions. If you reconnect, you will need to subscribe again.
• This does not affect other namespaces (e.g., jobs or triggers).

The _update event provides feedback when a subscription is successfully stopped.

socket.on('_update', (message) => {
    console.log("Update:", message);
    // Update: Stopped listening to vehicle-events for vehicle 617
});

If your application closes or disconnects from the server, it's good practice to clean up your subscriptions.

socket.on('disconnect', () => {
    console.log('Disconnected from Pegasus Live server. Cleaning up subscriptions...');
});

Payload Breakdown

The payload varies according to namespace.

vehicle-events

Received whenever a vehicle sends a tracking event.

The payload has a lot of keys, but the most important/relevant ones are:

• device - contains information about the device, like connection/network status, trip info, state of the Inputs/Outputs
• event - contains the values of the event keys reported
• primary - info on the primary object/entity (vehicle, asset, etc)
• secondary - info on the secondary object/entity (asset, bluetooth tag, etc)
• updates - tells you what key updated (the event, or device, or both)

As mentioned above the primary and secondary keys correspond to the vehicle or asset that reported the value.
For example, when a driver (an asset) is identified via an iButton with the vehicle, the primary would be the vehicle, and secondary would be the asset.
When the secondary_id is null it means that the secondary info shows the last active info.
In other words, if the secondary_id shows a value it means the secondary is current, if there's no value in secondary_id the secondary is the last known.
So this would be useful for knowing if a vehicle currently has an asset assigned, if so, which one, and if not, which was the last asset assigned.

The payload keys and values are the same as the one's you will expect when consulting via Rawdata

Please note that the same payload may be received twice, what you have to keep track of is the updates key, because it will tell you if it's a device update such as a connection state change (from online to offline for example) or an event update, which means that a new event came in.
On your applications you'll want to show only when a new event came in, and handle the connection states when the device key is received.

Note that keys or fields that begin with $$ or an underscore _ are reserved for internal use.

Payload for vehicle-events

{
    "$$hashKey": "object:1229",
    "category": "vehicle",
    "device": {"ios_state": {…}, "gpsknit": {…}, "network": {…}, "lastrx": {…}, "trip_setup": {…}, …},
    "event": {"code": 12, "gpsknit_prev_lat": 3257884, "sv": 8, "dev_dist__mi": 3899, "event_time": 1504909832, …},
    "pid": 1,
    "primary": {"info": {…}, "associations": Array(3), "__updated": 1504110100.755031, "name": "Syrus 3 SIENNA", "trackers": Array(0), …},
    "primary_id": 686,
    "secondary_id": null,
    "type": "events",
    "updates": ["device"],
    "_pkey": "pka.05",
    "_site_url": "https://cloud.pegasusgateway.com",
    "_source": "pcore.listener",
    "_source_epoch": 1504910674.863004,
    "_ver_core": "1.9.0-rc1",
    "_ver_pcel": "2.1.1-r2"
}

Device

> accel_state

Acceleration state

> asset

> config

configuration description

> connection

connection state / information

> dcounters

device counters

for a detailed description you can check the counters description

> gpsknit

gps information

> imei

IMEI is the imei of the device

> ios_state

state of the inputs and outputs

for a detailed description you can check the master field list

> lastrx

last communication time

> latest

latest data reported by the device

> lcounts

Listener counts, more information coming soon

> legacy_set_out_state

current state of the outputs, useful for knowing if output is pending to be set/reset

> lrates

Listener rates, more information coming soon

> muted_evs:

Muted or silenced events on the device information

> net_reg

Network registration information

> network

network information from device
Key | Description
----|-------------
cnum | carrier sim phone number (doesn't always reflect the number of the sim card due to restrictions from some carriers)
imsi | sim imsi
op | operator
sim | sim id
_epoch | epoch timestamp

> outbox

Array with the commands pending to be sent to the device

> photos

state of the cameras connected to the device

> position_response

Information about the latest response to a position message while the device was offline.

> satcom

Last satcom related information

> safeimmo_state

current state of the safe immobilization

> set_out_state

Output activation information, whether or not an instruction is pending to be sent to manipulate an output

> sigfox

Last sigfox related information

> sim

SIM card related information including iccid, imsi, operator, and more.

> trip

device trip information

> trip_session

device trip session information

> trip_setup

> vcounters

user editable vehicle counters, for more information go to counters

> vehicle

vehicle information

> version

> virtual_distance

virtual distance calculated via lat lon deltas

> virtual_ignition

virtual ignition calculated based on the status of the io_ign

Event

list of event parameters reported by device, check out master field list for more info

FAQ

Device Network State

The device network state can be found under device.connection.online, true whenever the device is ONLINE.

Valid Position

There are 3 keys that can be used to find out whether the position of the event is valid, they're all essentially the same:
device.data.valid_gps also in the event's keys: event.valid_gps
device.data.valid_position also in the event's keys: event.valid_position
device.latest.loc.valid

Authentication Errors

Try updating the socket.io client to version 2.0.3 or higher.