Notification API (1.0.0)

This API provides real-time notification delivery for voyage optimization services.

WebSocket is a communication protocol that provides full-duplex communication channels over a single TCP connection. Unlike traditional HTTP requests that follow a request-response pattern, WebSocket enables real-time, bidirectional communication between client and server.

Key Benefits of WebSocket:

• Real-time Communication: Instant message delivery without polling
• Low Latency: Minimal overhead for message transmission
• Persistent Connection: Single connection remains open for continuous communication
• Bidirectional: Both client and server can initiate message sending
• Efficient: Reduces server load compared to frequent HTTP polling

How it Works:

1. Client initiates a WebSocket handshake via HTTP upgrade request
2. Server accepts the upgrade and establishes a persistent WebSocket connection
3. Both parties can send messages at any time over the established connection
4. Connection remains open until explicitly closed by either party

This notification service delivers real-time notifications when reports are generated using Voyage Configuration API:

• Route Advice Reports
• Route Comparison Reports

You will receive WebSocket notifications once any of these reports are ready for download via the Product API.

Below are code examples showing how to connect to the WebSocket endpoint in different programming languages. These examples include:

• JWT token acquisition using client credentials flow
• WebSocket connection with proper authentication headers
• Automatic reconnection logic with retry strategies
• Token expiration handling and automatic token refresh
• Error handling for various disconnection scenarios

Node (JavaScript)

const WebSocket = require("ws");
const { URLSearchParams } = require("url");

class WebSocketClient {
  constructor() {
    this.token = null; // Will be obtained via OAuth2
    this.url = "wss://stg.api.voyageoptimization.abb.com/voyage/notification/v1/ws/products";
    this.ws = null;
    this.reconnectDelay = 5000; // 5 seconds
    this.shouldReconnect = true;
  }

  async connect() {
    console.log("Getting fresh token before connecting...");

    try {
      await this.refreshToken();
      console.log("Token obtained. Connecting to WebSocket...");
    } catch (error) {
      console.error("Failed to get token:", error.message);
      return;
    }

    this.ws = new WebSocket(this.url, {
      headers: {
        Authorization: `Bearer ${this.token}`,
      },
    });

    this.ws.on("open", () => {
      console.log("WebSocket connection established");
    });

    this.ws.on("message", (data) => {
      try {
        const notification = JSON.parse(data.toString());
        console.log("Received notification:", notification);
      } catch (error) {
        // Unable to parse the data
        console.log("Received raw message:", data.toString());
      }
    });

    this.ws.on("close", (code, reason) => {
      console.log(
        `WebSocket connection closed: ${code} - ${reason.toString()}`
      );

      // If close code is 4003, token is expired - automatically refresh token
      if (code === 4003) {
        console.log("Token expired (4003). Attempting to refresh token...");
        this.refreshToken()
          .then(() => {
            console.log("Token refresh successful. Reconnecting...");
            if (this.shouldReconnect) {
              setTimeout(() => this.connect(), 1000);
            }
          })
          .catch((error) => {
            console.error("Token refresh failed:", error.message);
            console.log(
              "Please check your OAuth2 credentials or refresh token manually"
            );
            if (this.shouldReconnect) {
              console.log(
                `Reconnecting in ${this.reconnectDelay / 1000} seconds...`
              );
              setTimeout(() => this.connect(), this.reconnectDelay);
            }
          });
        return;
      }

      if (this.shouldReconnect) {
        console.log(`Reconnecting in ${this.reconnectDelay / 1000} seconds...`);
        setTimeout(() => this.connect(), this.reconnectDelay);
      }
    });

    this.ws.on("error", (error) => {
      console.error("WebSocket error:", error.message);
    });
  }

  updateToken(newToken) {
    this.token = newToken;
    console.log("Token updated");
  }

  // Get a new access token using OAuth2 client credentials flow
  async refreshToken() {
    try {
      const response = await fetch(
        "https://internal.identity.genix.abilityplatform.abb/public/api/oauth2/token",
        {
          method: "POST",
          headers: {
            "Content-Type": "application/x-www-form-urlencoded",
            Accept: "application/json",
          },
          body: new URLSearchParams({
            grant_type: "client_credentials",
            client_id: "client_id",
            client_secret: "client_secret",
            scope: "https://genb2crs03euwprod.onmicrosoft.com/rs.iam/region",
          }),
        }
      );

      if (!response.ok) {
        const errorData = await response
          .json()
          .catch(() => ({ error: "Unknown error" }));
        throw new Error(
          `Token refresh failed: ${errorData.error || response.statusText}`
        );
      }

      const data = await response.json();
      this.token = data.access_token;
      console.log("Token refreshed successfully");
      return data.access_token;
    } catch (error) {
      console.error("Token refresh request failed:", error);
      throw error;
    }
  }

  disconnect() {
    this.shouldReconnect = false;
    if (this.ws) {
      this.ws.close();
    }
  }

  start() {
    this.shouldReconnect = true;
    this.connect();
  }
}

// Create and start the client
const client = new WebSocketClient();
client.start();

// Graceful shutdown on SIGINT (Ctrl+C)
process.on("SIGINT", () => {
  console.log("\nReceived SIGINT. Shutting down gracefully...");
  client.disconnect();
  process.exit(0);
});

// Example: How to manually refresh token
// client.refreshToken().then(() => console.log("Manual token refresh successful"));

// Example: How to manually update token
// client.updateToken("your-new-jwt-token-here");

Python

import asyncio
import websockets
import json
import aiohttp
from urllib.parse import urlencode

class WebSocketClient:
    def __init__(self):
        self.token = None  # Will be obtained via OAuth2
        self.uri = "wss://stg.api.voyageoptimization.abb.com/voyage/notification/v1/ws/products"
        self.reconnect_delay = 5  # 5 seconds
        self.should_reconnect = True
        
        # OAuth2 credentials
        self.client_id = "client_id"
        self.client_secret = "client_secret"
        self.scope = "https://genb2crs03euwprod.onmicrosoft.com/rs.iam/region"
        self.token_endpoint = "https://internal.identity.genix.abilityplatform.abb/public/api/oauth2/token"

    async def refresh_token(self):
        """Get a new access token using OAuth2 client credentials flow"""
        try:
            data = {
                'grant_type': 'client_credentials',
                'client_id': self.client_id,
                'client_secret': self.client_secret,
                'scope': self.scope
            }
            
            async with aiohttp.ClientSession() as session:
                async with session.post(
                    self.token_endpoint,
                    headers={'Content-Type': 'application/x-www-form-urlencoded'},
                    data=urlencode(data)
                ) as response:
                    if response.status == 200:
                        result = await response.json()
                        self.token = result['access_token']
                        print("Token refreshed successfully")
                        return self.token
                    else:
                        error_data = await response.json()
                        raise Exception(f"Token refresh failed: {error_data.get('error', 'Unknown error')}")
        except Exception as e:
            print(f"Token refresh request failed: {e}")
            raise e

    async def connect(self):
        """Connect to WebSocket with fresh token"""
        print("Getting fresh token before connecting...")
        
        try:
            await self.refresh_token()
            print("Token obtained. Connecting to WebSocket...")
        except Exception as e:
            print(f"Failed to get token: {e}")
            raise e  # Let start() method handle retry

        headers = {"Authorization": f"Bearer {self.token}"}
        
        try:
            async with websockets.connect(self.uri, additional_headers=headers) as websocket:
                print("WebSocket connection established")
                
                async for message in websocket:
                    try:
                        notification = json.loads(message)
                        print(f"Received notification: {notification}")
                    except json.JSONDecodeError:
                        print(f"Received raw message: {message}")
                        
        except websockets.exceptions.ConnectionClosedError as e:
            print(f"WebSocket connection closed: {e}")
            
            # Check close code for token expiration (4003)
            if hasattr(e, 'code') and e.code == 4003:
                print("Token expired (4003). Attempting to refresh token...")
                try:
                    await self.refresh_token()
                    print("Token refresh successful.")
                except Exception as error:
                    print(f"Token refresh failed: {error}")
                    print("Please check your OAuth2 credentials")
            
            # Let the start() method handle reconnection
            
        except Exception as e:
            print(f"WebSocket error: {e}")
            # Let the start() method handle reconnection

    def update_token(self, new_token):
        """Update token manually"""
        self.token = new_token
        print("Token updated")

    def disconnect(self):
        """Stop reconnection attempts"""
        self.should_reconnect = False

    async def start(self):
        """Start the WebSocket client with continuous reconnection"""
        self.should_reconnect = True
        
        while self.should_reconnect:
            try:
                await self.connect()
                # If we get here, connection was lost, wait before reconnecting
                print(f"Reconnecting in {self.reconnect_delay} seconds...")
                await asyncio.sleep(self.reconnect_delay)
            except KeyboardInterrupt:
                print("\nReceived interrupt. Shutting down gracefully...")
                self.should_reconnect = False
                break
            except Exception as e:
                print(f"Connection failed: {e}")
                print(f"Retrying in {self.reconnect_delay} seconds...")
                await asyncio.sleep(self.reconnect_delay)

async def main():
    print("Starting WebSocket client with reconnection...")
    client = WebSocketClient()
    
    try:
        await client.start()
    except KeyboardInterrupt:
        print("\nReceived interrupt. Shutting down gracefully...")
        client.disconnect()

if __name__ == "__main__":
    asyncio.run(main()) 

C#

using System;
using System.Net.WebSockets;
using System.Text;
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;
using System.Net.Http;
using System.Collections.Generic;

public class WebSocketClient
{
  private ClientWebSocket? _webSocket;
  private CancellationTokenSource? _cancellationTokenSource;
  private string? _token;
  private readonly string _uri = "wss://stg.api.voyageoptimization.abb.com/voyage/notification/v1/ws/products";
  private readonly int _reconnectDelay = 5000; // 5 seconds
  private bool _shouldReconnect = true;

  // OAuth2 credentials
  private readonly string _clientId = "client_id";
  private readonly string _clientSecret = "client_secret";
  private readonly string _scope = "https://genb2crs03euwprod.onmicrosoft.com/rs.iam/region";
  private readonly string _tokenEndpoint = "https://internal.identity.genix.abilityplatform.abb/public/api/oauth2/token";

  public async Task<string> RefreshTokenAsync()
  {
    using var httpClient = new HttpClient();

    var parameters = new List<KeyValuePair<string, string>>
    {
      new("grant_type", "client_credentials"),
      new("client_id", _clientId),
      new("client_secret", _clientSecret),
      new("scope", _scope)
    };

    var content = new FormUrlEncodedContent(parameters);

    try
    {
      var response = await httpClient.PostAsync(_tokenEndpoint, content);

      if (response.IsSuccessStatusCode)
      {
        var responseContent = await response.Content.ReadAsStringAsync();
        var tokenResponse = JsonSerializer.Deserialize<JsonElement>(responseContent);
        _token = tokenResponse.GetProperty("access_token").GetString();
        Console.WriteLine("Token refreshed successfully");
        return _token ?? throw new Exception("Token is null after refresh");
      }
      else
      {
        var errorContent = await response.Content.ReadAsStringAsync();
        throw new Exception($"Token refresh failed: {errorContent}");
      }
    }
    catch (Exception ex)
    {
      Console.WriteLine($"Token refresh request failed: {ex.Message}");
      throw;
    }
  }

  public async Task ConnectAsync()
  {
    Console.WriteLine("Getting fresh token before connecting...");

    try
    {
      await RefreshTokenAsync();
      Console.WriteLine("Token obtained. Connecting to WebSocket...");
    }
    catch (Exception ex)
    {
      Console.WriteLine($"Failed to get token: {ex.Message}");
      throw;
    }

    _webSocket = new ClientWebSocket();
    _webSocket.Options.SetRequestHeader("Authorization", $"Bearer {_token}");
    _cancellationTokenSource = new CancellationTokenSource();

    try
    {
      await _webSocket.ConnectAsync(new Uri(_uri), _cancellationTokenSource.Token);
      Console.WriteLine("WebSocket connection established");

      await ReceiveMessages();
    }
    catch (WebSocketException ex) when (ex.WebSocketErrorCode == WebSocketError.ConnectionClosedPrematurely)
    {
      Console.WriteLine($"WebSocket connection closed: {ex.Message}");
    }
    catch (Exception ex)
    {
      Console.WriteLine($"WebSocket error: {ex.Message}");
    }
  }

  private async Task ReceiveMessages()
  {
    var buffer = new byte[1024 * 4];

    try
    {
      while (_webSocket?.State == WebSocketState.Open)
      {
        var result = await _webSocket.ReceiveAsync(
            new ArraySegment<byte>(buffer),
            _cancellationTokenSource?.Token ?? CancellationToken.None
        );

        if (result.MessageType == WebSocketMessageType.Text)
        {
          var message = Encoding.UTF8.GetString(buffer, 0, result.Count);
          try
          {
            var notification = JsonSerializer.Deserialize<object>(message);
            Console.WriteLine($"Received notification: {notification}");
          }
          catch (JsonException)
          {
            Console.WriteLine($"Received raw message: {message}");
          }
        }
        else if (result.MessageType == WebSocketMessageType.Close)
        {
          Console.WriteLine($"WebSocket connection closed: {result.CloseStatus} - {result.CloseStatusDescription}");
          break;
        }
      }
    }
    catch (OperationCanceledException)
    {
      Console.WriteLine("WebSocket operation was cancelled");
    }
    catch (WebSocketException ex)
    {
      Console.WriteLine($"WebSocket connection closed: {ex.Message}");
    }
  }

  public void UpdateToken(string newToken)
  {
    _token = newToken;
    Console.WriteLine("Token updated");
  }

  public void Disconnect()
  {
    _shouldReconnect = false;
    _cancellationTokenSource?.Cancel();
  }

  public async Task StartAsync()
  {
    _shouldReconnect = true;

    while (_shouldReconnect)
    {
      try
      {
        await ConnectAsync();
        // If we get here, connection was lost, wait before reconnecting
        Console.WriteLine($"Reconnecting in {_reconnectDelay / 1000} seconds...");
        await Task.Delay(_reconnectDelay);
      }
      catch (Exception ex)
      {
        Console.WriteLine($"Connection failed: {ex.Message}");
        Console.WriteLine($"Retrying in {_reconnectDelay / 1000} seconds...");
        await Task.Delay(_reconnectDelay);
      }
    }
  }
}

class Program
{
  static async Task Main(string[] args)
  {
    Console.WriteLine("Starting C# WebSocket client...");

    var client = new WebSocketClient();
    await client.StartAsync();

    Console.WriteLine("Press any key to exit...");
    Console.ReadKey();
  }
}

Important Notes:

• Authentication: All connections require a valid JWT token in the Authorization header
• URL: Use wss:// for secure WebSocket connections
• Error Handling: Always implement proper error handling and reconnection logic
• Message Format: All messages are JSON-formatted strings

The WebSocket server implements automatic connection health monitoring:

Heartbeat Mechanism

• Ping Interval: Server sends ping frames every 30 seconds
• Pong Requirement: Clients must respond with pong frames
  • Automatic Handling: Most WebSocket libraries (including those used in the code examples above) automatically handle pong responses to ping frames
  • No Manual Implementation: You typically don't need to manually implement pong logic - the library handles it transparently
• Missed Pongs: If client fails to respond to 10 consecutive pings, connection is terminated
• Auto-cleanup: Unresponsive connections are automatically removed

WebSocket Close Codes

The server uses specific close codes to indicate different disconnection reasons:

Important for Client Implementation:

• Monitor close codes in your WebSocket event handler
• Implement different retry strategies based on close code:
  • 1001: Retry after a short delay (server restart)
  • 1008: Retry after a longer delay (server capacity)
  • 4003: Refresh token before reconnecting (authentication issue)

Reconnection Scenarios

Clients must handle reconnection for the following situations:

• Network connectivity issues (temporary network outages)
• Server maintenance (planned or unplanned service restarts)
• Token expiration (close code 4003 - requires token refresh before reconnecting)
• Heartbeat failures (connection is immediately terminated after 10 missed pongs)
• Connection limits (server at capacity)
• Unexpected errors (server-side exceptions, protocol errors)

Authentication endpoints

Service health monitoring endpoints

Real-time WebSocket communication endpoints for product notifications

The Notifications API sends announcements for new routes, route advices and comparison reports created by the Voyage Configuration API.

The notifications from the Notification API will refer to a download URL on the Product API.