DEV Community: kaul-vineet

Crypto Exchange: Supercharge customer apps with streaming data.

kaul-vineet — Wed, 25 Nov 2020 07:53:30 +0000

Team: Vineet Kaul, Ajay Kumar

What are we talking about?

We are talking about streams. Streams of data originating from applications, social media, sensors, devices, websites, blah, blah, blah. The great book of Wikipedia describes streaming data as ...data that is continuously generated by different sources. Streaming data is ubiquitous and if you've never seen it in action, think again...

Sensors in transportation vehicles, industrial equipment, and farm machinery.
Stock market data, portfolios, and value-at-risk calculation in real-time.
Clickstream records from digital content.
Data about player-game interactions in online gaming.

All this sounds interesting, but managing streaming data does not come without its fair share of challenges. Broadly speaking,

Processing: Streaming data requires real-time data pipelines for managing the flow; correlations, aggregations, filtering, and sampling.
Storage: Business requires a view to the data, aggregated or over sliding time windows. But it is commercially prohibitive to store everything as the growth is algorithmic.
Display: Business needs an aggregated window view but it can't be saved, so how to view it? Hence a need for a dynamic unified interface to give a view window on data.

Houston, do we have a problem?

Not exactly. There is a way to supercharge customer apps with streaming data by using the technology components that take care of the underlying complexities so developers can manage on delivering value to the business.

Business use case:

Personae:

Art Vandalay is an investor with interests in cryptocurrency trading. Art trades on various crypto exchanges across various cryptocurrencies. Art is also a customer of Chemical Bank and would prefer if his banking app would help him track cryptocurrency exchange rates.

Chemical Bank is just about to that with their app Crypto Exchange.

Cryptocurrency trade basics:

Source: Investopedia
Trading cryptocurrencies requires a currency pair. In other words, the value of one currency being quoted against the other. The first listed currency of a currency pair is called the base currency, and the second currency is called the quote currency. It indicates how much of the quote currency is needed to purchase one unit of the base currency. Currencies are identified by an ISO currency code.

The trading rates of cryptocurrencies change in real-time. To develop Crypto Exchange, Chemical Bank would need to hook up to an exchange, ingest streaming data on exchange rates of hundreds of pairs, and display it in a dynamic unified interface for Mr. Vandalay's trading pleasure.

Chemical Bank is going to do just that using Salesforce technology stack.

Architecture:

The solution takes CoinCap as a source of cryptocurrency trading data. Why? Just convenient. It could be any of many data sources over the internet streaming exchange data.

Commentary:

Mulesfot WebSockets Connector connects to the CoinCap WebSocket. The CoinCap trade WebSocket streams exchange rate data from other cryptocurrency exchange WebSockets. Because the system is designing for streaming data and not API callouts, Mulesoft WebSockets Connector opes & maintains the TCP connection between a client (Crypto Exchange) and server (CoinCap) until the app closes it.
The streaming data from any source could be overwhelming. Downstream systems might not be able to process or store data at a very high rate. The design alternative is to aggregate or throttle the data as long as it does not affect the business use case. Mulesoft Aggregators Module stores the exchange values and releases the most recent message based on the configuration (count or time). As long as the lag is acceptable, the downstream flow could be optimized. Remember, Mr. Vandalay just wants to track cryptocurrency exchange rates, and not trade through the app. The current strategy does not implement aggregation in principle but throttling. That said, the same strategy could be used to implement other use cases that require aggregation e.g. calculating the mean of exchange rates over a time duration.
Post-processing, Mulesoft sends the data for display to downstream app i.e. Crypto Exchange. As there is neither need nor technical feasibility to store data, the app would graphically display the exchange data. Mulesoft sends a Salesforce platform event to the app that is subscribed to by the UI components. The lightning:empApi component receives the event and displays it on Chart.js graphs. The use of Salesforce Platform events simplifies communication by decoupling event producers from aggregator and subscriber. In addition, it is possible for multiple apps to subscribe to a single solution.

And in three simple steps, we have an end to end flow that connects to a crypto exchange and displays data on a customer app.

Mulesoft:

The Listener: On New Inbound Message listens to the connection requests and calls the flow to open the connection through Open outbound socket.

The Aggregator: Group based aggregator collects the exchange rates based on a combination of base and quote currencies and releases the latest rate every "x" messages (where "x" could be any suitable batch size). This scheme is called Size-based aggregation. Similar to the approach, Time-based aggregation could be designed as well.

Once again, this is done to throttle the message stream so as to make it easier for downstream systems to ingest and process the exchange rate messages. Mulesoft orchestration imposes a sliding window on the data stream of which on one and the latest message is sent downstream. This chunking of data enables the architecture to accommodate customer-facing applications that are either not designed for streaming data or have platform governance limitations.

A single pair message:

{exchange=binance, base=litecoin, quote=tether, direction=buy, price=68.03, volume=12.82018, timestamp=1606478960197, priceUsd=68.14666514408307}

Aggregated message for throttle. Only the latest message in a time/count window would be passed.

[TypedValue[value: '{exchange=binance, base=litecoin, quote=tether, direction=buy, price=67.65, volume=1.16843, timestamp=1606479662239, priceUsd=67.75263835061388}', dataType: 'MapDataType{type=java.util.LinkedHashMap, keyType=SimpleDataType{type=java.lang.Object, mimeType='/'}, valueType=SimpleDataType{type=java.lang.Object, mimeType='/'}, mimeType='application/java; charset=UTF-8'}']]

The Publisher: Once the throttled messages are released, Salesforce connector publishes a Salesforce platform event to the connected Org. A custom platform event definition is described in the connected Org.
Platform events are built for handling streaming data. Platform events also allow for the processing and display of data without storing the data and hence avoiding costly DB operations.

Salesforce:

The Consumer: Display of data on dynamic graphs requires the UI component to receive the published messages. This is implemented by the lightning:empApi component that provides access to methods for subscribing to a streaming channel and listening to event messages. The lightning:empApi component uses a shared CometD connection, enabling dynamic display of streaming data on UI components.

Watch it work:

Can I get you the code? Yeah, sure.

So, you are not faint-hearted. Fair enough. We have published the code and full setup-up and test instructions on GitHub. Follow the lead.

For the Salesforce app code and detailed instructions to set up and test the hack, refer to the repo below. This covers the full hack i.e. Salesforce & Mulesoft part.

kaul-vineet / StreamingCurrency-sf

Crypto Exchange

How to integrate you Salesforce application with streaming data from Crypto Exchange.

Part 1: Setup the Salesforce Org.

Click on the button below to install the DriveSafe application package. The Salesforce Org should be Enterprise and above.

Import seed data into the Org
- Use the plan file and data file to load data into Trade_Asset__c object in the Org. Use the sf:dx command to import data:

sfdx force:data:tree:import -p data/Trade_Asset__c-plan.json -u <Org Alias>

Another alternative is to convert data file to CSV and upload the Trade_Asset__c.Name field by using DataLoader, Workbench or your favorite data management tool
Open Salesforce1 app and go to App Launcher.

Search and select Crypto Exchange app.

Go to trades and click on the New button.

A form appears for adding a new trade. Select the Base Asset and Quote Asset.

If Trade_Asset__c data has been properly uploaded, options would appear.

Any possible…

View on GitHub

If you are interested in the Mulesoft project, refer to the repo below.

kaul-vineet / StreamingCurrency-mule

StreamingCurrency-mule

View on GitHub

Have fun trying out the hack and do let us know if you face issues or general feedback.

🚖 DriveSafe: An IoT use case for auto industry 🚖

kaul-vineet — Mon, 23 Nov 2020 10:52:25 +0000

📟📟 COLLECT BUSINESS INTELLIGENCE FROM PHONE OR ANY IoT ENABLED DEVICE LEVERAGING SALESFORCE PLATFORM. 📟📟

Team: Vineet Kaul, Ajay Kumar, Alekhya

IoT Is Mainstream.

All technology professionals are increasingly being inundated with customer requirements to connect devices to CRM and/or customer engagement solutions. It is not just important to connect devices with CE systems, but the systems should ingest the streaming data and produce actionable insights into business behavior. IoT today is no longer technology, but a business problem and requires a business-centric solution. Today, companies use IoT data for operational performance in internal silos.

While the goals are simple, orchestrating such a system is could be an architectural mix of multiple technologies. By using the Salesforce technology stack, architects can design and deliver a seamless system that translates IoT data streams to data intelligence driving business decisions.

One way to accomplish this That is what we have tried to achieve in this submission.

The post takes two business scenarios and provides a detailed description of one of the many technology solutions to address them. For the more adventurous readers, it provides a description of code (GitHub repository included) and a self assist guide to implement a technology demonstrator on their own.

Uses In The CRM Industry.

🚑 Emergency Services: Track sudden change in altitudes of cars outside of safe envelop to detect accidents.

🚫 Insurance: Track driving behavior to determine insurance claims and renewal benefits.

📢 Marketing: Track gradual changes in altitude to send contextual notifications.

🚙 Automotive: Track vehicle acceleration, orientation to determine service plan.

And other amazing things you can think of !!!!!!

Business Scenarios

Usecase # 1: The same insurance company provides value-added services like emergency assistance and pick up to its insurance holders. The errant drivers for one would certainly need it.

In case of an emergency, the customer engagement application would track beacon messages from IoT-enabled devices to get the real-time location (static or kinetic) of the vehicle. While the application gets the immediate coordinates of the distressed vehicle(static), it also tracks the beacons of the moving vehicle. A combination of these could be used for emergency response or guiding drivers to the nearest workshop or gas station. And yes, you've read it right, everything is real-time.

Usecase # 2: An insurance company wants to link premium with driving behavior. It provides its customers with a device or mobile-based app that is kept on while driving. The solution pings sensor output (linear X velocity) to the insurance application.

If the speed exceeds a threshold value (let's say 20 miles/hr), the platform event engine automatically creates a "Violation" record with violation details i.e. excess speed, location, etc (yes, everything is real-time). The violation is attached to the customer policy record and each violation results in an additional surcharge at the time of policy renewal. Or in other words, no violation means no surcharge and less costly insurance renewal.

Technology & Architecture.

While it looks like complex architecture, with Salesforce technology stack, this could be orchestrated with clicks not code. Below, we describe the moving parts of the solution in detail.

MOBILE:
A simple app is required that sends a stream of sensor values as the customer is driving. MQTT is the best protocol for sending lightweight messages from a mobile device. In real life scenario, mobile could be replaced with an in-car appliance, satellite tracker, or anything that has a battery and connects to the net. In the solution, we transmit a stream of linear X velocity of the mobile phone to simulate vehicle speed.

HEROKU:
*CloudMQTT: The streams of sensor pings require a broker. This is provided by CloudMQTT, a Heroku based message broker. The broker receives the message stream that any customer based system can subscribe to. Such a pub-sub architecture has the following advantages

It de-couples the devices with customer engagement systems. Any new system can just plug into the stream rather than being hard-wired to it.
It allows message streams to be published under different topics e.g. speed, tilt, acceleration, etc. In the case of multiple systems, each customer engagement system could subscribe to the topic based on its business requirements.

Node-RED + MULESOFT
Once the MQTT broker receives a message, it is to be consumed and routed. This is termed as IoT orchestration. A combination of
Node-RED & Mulesoft flow forms the ESB backbone that does the following:

Node-RED: Integrates with the MQTT broker and collects the message stream. It throttles the message stream to ensure that downstream systems are not overwhelmed by the data overflows. Many customer engagement systems have data ingestion limits and strategies like throttling ensure downstream stability. There is no business sense as well to create multiple violation records for the same speeding instance, is it? Node-RED also strips the complex JSON message and creates a simpler message with limited values.

Mulesoft: Works as a messaging glue between the message stream and the subscriber systems. Mulesoft is the ESB backbone that publishes the message stream as a Salesforce platform event (we assume for this solution that our customer engagement platform is built on Salesforce). Multiple Salesforce Orgs can subscribe to the message and use it for their individual business logic. Mulesoft provides an inbuilt connector that publishes custom or generic platform events described in Salesforce. The advantages of an event-based approach are

Events order of delivery for a specific channel is guaranteed.
Event subscribers can be defined and integrated with the platform business logic through clicks not code.
Platform Events are non-transactional thereby resulting in no database overhead (an important factor for multi-tenant systems) and have high volume support.

SALESFORCE PLATFORM:
(we assume for this solution that our customer engagement platform is built on Salesforce)

Usecase #1: Everything revolves around the subscription of platform events published by Mulesoft. For this use case, an empApi aura component displays the real-time message stream from Mulesoft. The component dynamically sets the map markers on receipt of each platform event thereby dynamically tracking the message source.
Usecase #2: Salesforce Flow subscribes to the specified platform event published by Mulesoft. When a platform event message with sensor event is received, the flow orchestrates and implements the business process i.e. check the speed value and create a "violation" record.

Watch the demonstration (Usecase # 1).

Watch the demonstration (Usecase # 2).

Want to give it a test drive?.

Complete installation and demo steps are documented in the README.md of repository
kaul-vineet / DriveSafe
DriveSafe: An IoT use case for auto industry.

Built and orchestrated on Salesforce platform.

All technology professionals are increasingly being inundated with customer requirements to connect devices to CRM and/or customer engagement solutions. It is not just important to connect devices with CE systems, but the systems should ingest the streaming data and produce actionable insights into business behavior. IoT today is no longer technology, but a business problem and requires a business-centric solution. Today, companies use IoT data for operational performance in internal silos.

That is what we have tried to achieve in this submission.

There are 4 parts to the setup, fron setup to test. They might look complicated but everything is point & click configuration.

Step 1: Deploy Salesforce application.
- Click on the button below to install the DriveSafe application package. The Salesforce Org should be Enterprise and above.
- On successful completion of the deployment, log-in into…
View on GitHub
.
A separate repository for Mulesoft flow could be referenced (this repository is not required for setup) at

kaul-vineet / DriveSafe-mule
There are 4 parts to the setup, fron setup to test. They might look complicated but everything is point & click configuration.
In case of trouble, post a comment on GitHub or this post.

Design Considerations:

As discussed in the architecture explanation, Platform Events are not unlimited. Even the High Volume Platform Events are governed by daily limits. For use cases where the message stream volume is very high, follow strategies that would limit the production & consumption of events and events.

Use throttle while delivering IoT events: Process each message in the message stream at a fixed rate. This rate could be based on the volume, size, or time. We have followed this strategy in this solution.
Use Report By Exception strategy: Block messages unless its value has changed.