/**
* The function that handles the call to the API when polling
*/
callEndpoint = async () => {
    const response = await fetch(this.apiUrl);
    const contentLength = response.headers.get("content-length");
    const data = await response.json();
    // Show the data from the API in the response element
    const formatter = new JSONFormatter(data);
    this.responseBodyEl.innerHTML = '';
    this.responseBodyEl.appendChild(formatter.render());
    
    // Publish polled data to Redis Websocket
    this.redisWebSocket.send(JSON.stringify(data));
}

server.on('connection', function connection(ws) {
    // Message Received in Socket from webclient, and sent to redis
    ws.on('message', function (event) {
        console.log('Message from Websocket:', event);
        redisPublisher.publish(REDIS_CHANNEL, event);
    });

    // Message received from Redis, and sent through Websocket to Webclients
    redisSubscriber.on('message', function (channel, message) {
        console.log('Redis Subscriber: ', channel, message);
        ws.send(message);
    })
});

/**
     * Add the websocket listener to listen for Redis Messages
     */
    startListeningRedisWebSocket = () => {
        console.log('--- BackendApp: Connecting to WebSocket ---');
        /* Then, we setup Redis Topic listener on the websocket */
        this.redisWebSocket.onmessage = ({ data }) => {            
            this.message = JSON.parse(data); // Parse the data from Redis
            console.log('Data received from Redis: ', this.message);
            
            // Publish received data to Diffusion
            this.publishToDiffusion(this.message);
        }
    }

    /**
     * Publish data received from socket to Diffusion
     * @param {*} data 
     */
    publishToDiffusion = data => {                
        if (this.diffusionService && this.diffusionService.diffusionClient) {
            this.diffusionService.publish(data);
        }
    }

/**
 * Subscribe to a diffusion Topic.
 * If it's not yet connected, first connect
 */
subscribeToDiffusion = () => {
    if (!this.diffusionService.diffusionClient) {
        console.log('*** Diffusion Client: Connecting to Diffusion ***');
        this.diffusionService.connect(
            this.hostEl.value,
            this.userEl.value,
            this.passwordEl.value,
            this.topic,
            message => this.onDiffusionMessage(message)
        );
    }
}

/**
 * This is the callback the Diffusion Client calls when a message is received
 * We update the Client Tier chart with this info
 * @param {*} message 
 */
onDiffusionMessage = message => {
    console.log('on Diffusion message', message);

    this.updateDataReceived(JSON.stringify(message).length);

    // This message came from Diffusion! Feed Diffusion's Chart
    this.updateChart(message);
}

/**
     * This is the event handler when the Connect to Diffusion button is clicked
     * @param {*} evt 
     */
    connect = (
        host, user, password, topic, onMessageCallback
    ) => {
        console.log('Connecting to Diffusion');        

        // Instantiate Diffusion's Client
        // We send the connect and on message callbacks to handle those events
        this.diffusionClient = new Diffusion(this.onConnectedToDiffusion, onMessageCallback || this.onDiffusionMessage);

        // Set Diffusion config
        this.diffusionClient.setConfig({
            host: host,
            user: user,
            password: password,
            topic: topic
        });

        // And connect to it
        this.diffusionClient.connect();
    }

    /**
     * This is the callback, Diffusion client calls after connection
     */
    onConnectedToDiffusion = () => {
        console.log('connected to diffusion');
        // Once we're connected, subscribe to the topic we specified when connecting to Diffusion service
        // We're not sending any parameters because we already set those when calling the setConfig function in the previous method.
        this.diffusionClient.subscribe({}); //Subscribe to Diffusion's topic
    }

And this is how the Service uses the SDK to publish messages:

/**
 * Publish to diffusion
 * @param {*} data 
 */
publish = data => {
    this.diffusionClient.publishData(data);
}

Diffusion SDK Client

Finally we have the piece of code in charge of directly interfacing with the Diffusion Javascript SDK. This is the class that’s used by the Diffusion Service in both Application and Client Tiers.

This is how it connects to Diffusion, using SDK functions:

/**
 * This method is used to configure Diffusion connection
 * @param host The host of the Diffusion Service
 * @param user The user to the Diffusion Service
 * @param password
 * @param topic The topic name to be created and consumer from
 */
setConfig = ({ host, user, password, topic }) => {         
    this.host = host || '127.0.0.1';
    this.user = user || 'admin';
    this.password = password || 'password';
    this.topic = topic || 'default-topic';
    this.subscribedTopic = topic || 'default-topic';
}

/**
 * Method that handles connection to the Diffusion Service      
 */
connect = () => {
    /* Connect to Diffusion using the parameters previously set in setConfig method */
    diffusion.connect({
        host: this.host,            
        principal: this.user,
        credentials: this.password,                       
        port: 443,
        secure: true
    }).then((session) => {
        this.session = session;
        /* Here's where we add the topic we are going to be using */
        this.session.topics.add(this.topic, diffusion.topics.TopicType.JSON)            
        console.log(`Connected: `, this.session.sessionId);

        /* If we setup a connected callback, let's call it */
        if (this.onConnectedCallback) {
            this.onConnectedCallback();
        }            
    });
}

This is how it subscribes to Topics, and handles messages received in that Topic:

/**
 * Method to subscribe to a topic and start consuming it
 * @param session We can pass an already existing session, otherwise it will use the internal one
 * @param topicPath Set a topic to subscribe to, if empty uses the internal one
 * @param onValueCallback callback to be called when a value is arrives in the topic. It can be null
 */
subscribe = ({ session = undefined, topicPath = '', onValueCallback = null }) => {
    const currentSession = session || this.session;
    const currentTopic = topicPath || this.topic;
    console.log(`subscribing to: ${currentTopic}`);
    
    /* We Setup the stream */
    currentSession.addStream(
        currentTopic,
        diffusion.datatypes.json()).on('value',
            onValueCallback || this.onReceiveMessage
        );

    /* And subscribe to the topic */
    currentSession.select(currentTopic);
}

onReceiveMessage = (topic, specification, newValue, oldValue) => {
    let message = newValue.get();
    message.receiveTime = new Date();
    console.log(`TOPIC - Receiving message for topic: ${topic}`, specification, newValue.get(), oldValue.get());
    if (this.onReceiveMessageCallback) {
        this.onReceiveMessageCallback(message, topic);
    }
}

And finally this is how it publishes messages to a Topic:

// Interface Functions
publishData(data) {        
    if (this.session) {
        this.session.topicUpdate.set(this.topic, diffusion.datatypes.json(), data);
    }
}

Pre-requisites

• Download our code examples or clone them to your local environment:

 git clone https://github.com/diffusion-playground/redis-integration/