Data Sharing
Overview
The Interop API is accessible through the io.interop
object.
See the JavaScript Interop Request/Response and Interop Streaming examples on GitHub.
Method Registration
To register an Interop method that will be available to all other interop-enabled apps, use the register()
method. Provide a name for the method (or a MethodDefinition
object) and a callback that will handle invocations from client apps:
// Required name for the method to register.
const methodName = "Addition";
// Required callback that will handle client invocations.
const handler = ({ a, b }) => {
const result = { sum: a + b };
return result;
};
await io.interop.register(methodName, handler);
After registration, the "Addition" Interop method will be available to all other interop-enabled apps and any of them will be able to invoke it with custom arguments at any time, as long the server offering it is running or until it unregisters it (with the unregister()
method).
Interop methods with the same name may be registered by different servers. An Interop method is considered the same as another Interop method if their names are the same and if the accepts
and returns
properties of their MethodDefinition
objects have identical values. The implementation of the handler function, however, may differ for each server.
Method Definition
When registering an Interop method, it is required to pass either a string for a method name or a MethodDefinition
object. The MethodDefinition
object describes the Interop method your app is offering. It has the following properties:
Property | Type | Description |
---|---|---|
accepts |
string |
Signature describing the arguments that the method expects (see Input and Output Signature). |
description |
string |
Description of the functionality the method provides. |
displayName |
string |
User-friendly name for the method that may be displayed in UIs. |
name |
string |
A name for the method. |
objectTypes |
string |
Predefined data structures (e.g., "Instrument" , "Client" , etc.) with which the method works (see Object Types). |
returns |
string |
Signature describing the return value of the method (see Input and Output Signature). |
supportsStreaming |
boolean |
Whether the method is an Interop stream. |
version |
number |
Method version. |
// Method definition.
const methodDefinition = {
name: "Addition",
accepts: "Int a, Int b, Int? c",
returns: "Int sum",
displayName: "Calculate Sum",
description: "Calculates the sum of the input numbers."
};
const handler = ({ a, b, c }) => {
const result = {
sum: a + b + (c ? c : 0)
};
return result;
};
await io.interop.register(methodDefinition, handler);
Input & Output Signature
Documenting the input and output method signatures is useful during development and testing. The input and output signatures are used in debugging tools like the Interop Viewer to show detailed information about the Interop method you are testing.
To describe the arguments that your Interop method expects and the value it returns, use the accepts
and returns
properties of the MethodDefinition
object. Both properties accept a comma-separated string of arguments. Each argument described in the string must use the following format:
type <array-modifier> <optional-modifier> argument-name (<description>)
// The `type` is one of:
type = "bool" | "int" | "double" | "long" | "string" | "datetime" | "tuple: {<schema>}" | "composite: {<schema>}"
// The `<schema>` represents any value(s) in the same format.
"Composite" is a structure which may contain one or more fields of scalar type, array of scalars, a nested composite or an array of composites. A "Composite" allows you to define almost any non-recursive structure.
Examples:
"string name, string[]? titles"
-name
is required,titles
is an optional string array;"tuple: { string name, int age } personalDetails"
-personalDetails
is a required tuple value containing two required values -name
as a string andage
as an integer;"composite: { string first, string? middle, string last } name"
-name
is a composite argument and its schema is defined by 2 required string fields -first
andlast
, and an optional string field -middle
;
Returning Results
When returning results from you Interop methods, wrap the return value in an object:
({ a, b }) => {
// Return an object.
return { sum: a + b };
};
Otherwise, the result will be automatically wrapped in an object with a single _value
property which will hold your return value:
({ a, b }) => {
// This will be automatically wrapped in an object.
return a + b;
};
// If a=2 and b=3, the resulting value will look like this:
// { _value: 5 }
Asynchronous Results
Interop methods can return asynchronous results as well. Use the register()
method to register an asynchronous Interop method:
const asyncMethodName = "MyAsyncMethod";
const asyncHandler = async () => {
const response = await fetch("https://docs.interop.io");
if (response.ok) {
return 42;
} else {
throw new Error("The doc site is down!");
};
};
await io.interop.register(asyncMethodName, asyncHandler);
Method Invocation
To invoke an Interop method, use the invoke()
method. The only required argument for invoke()
is a method name or a MethodDefinition
object. You can also specify arguments, target and other invocation options:
const methodName = "Addition";
const args = { a: 2, b: 3 };
const target = "all";
const options = {
waitTimeoutMs: 5000,
methodResponseTimeoutMs: 8000
};
const result = await io.interop.invoke(methodName, args, target, options);
Targeting
If multiple apps offer the same Interop method, you can choose to invoke it on the "best"
app instance (default), on a specific Interop instance, on a set of instances, or on all instances.
The following table describes the accepted target values when invoking an Interop method:
Value | Description |
---|---|
"all" |
Executes the method on all Interop servers offering it. |
"best" |
Default. Executes the method on the best (first) server (the io.Connect runtime determines the appropriate instance). |
"skipMine" |
Like "all" , but skips the current server. |
Instance |
An object describing an Interop instance. It is also possible to provide only a subset of the Interop instance object properties as a filter - e.g., { application: "appName" } . |
Instance[] | Array of Interop Instance objects (or subset filters). |
⚠️ Note that the properties of an Interop Instance object accept both a string or a regular expression as values.
App instances are ranked internally. The "best"
instance is the first one running on the user's desktop and under the user's name. If there are multiple apps matching these criteria, the first instance is used.
To invoke a method on a preferred set of apps, pass a target as a third argument.
To target all Interop instances offering the same method:
const target = "all";
await io.interop.invoke("Addition", { a: 2, b: 3 }, target);
To target all instances, except the current one:
const target = "skipMine";
await io.interop.invoke("Addition", { a: 2, b: 3 }, target);
To target a specific instance:
const target = { application: "Calculator" };
await io.interop.invoke("Addition", { a: 2, b: 3 }, target);
To target a set of instances (for more information on finding Interop instances, see Discovery):
const targets = io.interop.servers()
.filter(server => server.application.startsWith("Calculator"));
await io.interop.invoke("Addition", { a: 2, b: 3 }, targets);
If nothing is passed, "best"
is default:
await io.interop.invoke("Addition", { a: 2, b: 3 });
Consuming Results
The invoke()
method is asynchronous and resolves with an InvocationResult
object. Use the returned
property of the InvocationResult
object to extract the returned result:
const invocationResult = await io.interop.invoke("Addition", { a: 2, b: 3 });
// The method returns an object with a `sum` property.
const sum = invocationResult.returned.sum;
Multiple Results
Invoking a method on multiple Interop instances produces multiple results. Use the all_return_values
property of the InvocationResult
object to obtain an array of all invocation results:
const invocationResult = await io.interop.invoke("Addition", { a: 2, b: 3 }, "all");
invocationResult.all_return_values
.forEach(result => console.log(result.returned.sum));
Object Types
Use the objectTypes
property of the MethodDefinition
when registering an Interop method to specify what predefined data structures the method expects - e.g., "Instrument"
, "Client"
, etc. Specifying the object types in a method definition is useful for determining at runtime the methods applicable to the currently handled object. For the object types to function in a generic manner, all apps must follow the same data format and pass the respective objects to the respective Interop methods.
To register a method with object type specifications:
const methodDefinition = {
name: "SetClient",
objectTypes: ["Client"]
};
const handler = (client) => {
console.log(client.id, client.name);
};
await io.interop.register(methodDefinition, handler);
To find all methods working with a specific object type:
const clientMethods = io.interop.methods()
.filter(method => method.objectTypes?.includes("Client"));
To invoke a method working with a specific object type:
const methodDefinition = {
name: "SetClient",
objectTypes: ["Client"]
};
await io.interop.invoke(methodDefinition);
Discovery
Use the Interop Viewer to monitor all registered Interop methods and streams. Invoke methods and subscribe to streams with custom arguments to observe the results.
Methods
To get a collection of all available Interop methods, use the methods()
method:
const allMethods = io.interop.methods();
To find a specific method or a set of methods, pass a string or a MethodFilter
object:
const methodFilter = { name: "Addition" };
const filteredMethods = io.interop.methods(methodFilter);
To find all methods of an Interop instance:
const instance = { application: "appName" };
const methods = io.interop.methodsForInstance(instance);
If you have a reference to an Interop instance, use its getMethods()
and getStreams()
methods:
// Get the current Interop instance of the app.
const myInstance = io.interop.instance;
// Get the Interop methods registered by the instance.
const methods = myInstance.getMethods();
// Get the Interop streams registered by the instance.
const streams = myInstance.getStreams();
Servers
To get a collection of all Interop servers, use the servers()
method:
const servers = io.interop.servers();
To find the servers offering a specific method, pass a MethodFilter
object:
const methodFilter = { name: "Addition" };
const serversForMethod = io.interop.servers(methodFilter);
If you have a reference to a Method
object, use its getServers()
method:
const method = io.interop.methods("Addition")[0];
const servers = method.getServers();
Interop Events
The Interop API offers means for notifying you when a method has been added/removed or when an app offering methods becomes available/unavailable. All methods for listening for events return an unsubscribe function. Use it to stop receiving event notifications.
To get notified when a method has been added for the first time by any app, use methodAdded()
:
const handler = (method) => {
console.log(`Method "${method.name}" was added.`);
};
io.interop.methodAdded(handler);
To get notified when a method has been removed from the last app offering it, use methodRemoved()
:
const handler = (method) => {
console.log(`Method "${method.name}" was removed.`);
};
io.interop.methodRemoved(handler);
To get notified when an app offering methods has been discovered, use serverAdded()
:
const handler = (instance) => {
console.log(`Interop server was discovered: "${instance.application}".`);
};
io.interop.serverAdded(handler);
To get notified when an app stops offering methods or is closed, use serverRemoved()
:
const handler = (instance) => {
console.log(`Interop server was removed: "${instance.application}".`);
};
io.interop.serverRemoved(handler);
To get notified every time a method is offered by any app, use serverMethodAdded()
. This event fires every time any app starts offering a method, while methodAdded()
fires only for the first app which starts to offer the method:
const handler = (info) => {
const serverName = info.server.application;
const methodName = info.method.name;
console.log(`Interop server "${serverName}" now offers method "${methodName}".`);
};
io.interop.serverMethodAdded(handler);
To get notified every time a method is removed from any app, use serverMethodRemoved()
. This event fires every time any app stops offering a method, while methodRemoved()
fires only when the method has been removed from the last app offering it:
const handler = (info) => {
const serverName = info.server.application;
const methodName = info.method.name;
console.log(`Interop server "${serverName}" has removed method "${methodName}".`);
};
io.interop.serverMethodRemoved(handler);
Streaming
Overview
Your app can publish events that can be observed by other apps and can provide real-time data (e.g., market data, news alerts, notifications, etc.) to other apps by publishing an Interop stream. It can also receive and react to these events and data by creating an Interop stream subscription.
Apps that create and publish to Interop streams are called "publishers", and apps that subscribe to Interop Streams are called "subscribers". An app can be both.
See the JavaScript Streaming example on GitHub.
The following snippets are based on the Complete Streaming Example at the end of the Streaming section. The example illustrates an app which retrieves real-time data about a financial instrument (symbol
) from a data source and provides the data as a stream to other apps. The client apps subscribe for the stream data by providing a symbol
value as an argument in their subscription requests.
Publishing Stream Data
Creating Streams
To start publishing data, create an Interop stream by calling createStream()
. This registers an Interop method similar to the one created by register()
, but with streaming semantics.The createStream()
method accepts a string or a MethodDefinition
object as a first argument and a StreamOptions
as a second.
The MethodDefinition
is identical to the Interop method definition for the io.interop.register()
method. If you pass a string, it will be used as a stream name:
const stream = await io.interop.createStream("MarketData.LastTrades");
Which is identical to:
const streamDefinition = { name: "MarketData.LastTrades" };
const stream = await io.interop.createStream(streamDefinition);
The StreamOptions
object allows you to pass several optional callbacks which let your app handle subscriptions in a more detailed manner:
- to identify individual subscribers/clients;
- to accept or reject subscriptions based on the subscription arguments;
- to unicast data as soon as a client subscribes to the stream;
- to group subscribers which use the same subscription arguments on a stream branch and then publish to that branch, multicasting data to all subscribers;
StreamOptions
object example:
const streamOptions = {
subscriptionRequestHandler: subscriptionRequest => {},
subscriptionAddedHandler: streamSubscription => {},
subscriptionRemovedHandler: streamSubscription => {}
};
Example of creating a stream:
// Stream definition.
const streamDefinition = {
name: "MarketData.LastTrades",
displayName: "Market Data - Last Trades",
accepts: "String symbol",
returns: "String symbol, Double lastTradePrice"
};
// Stream options object containing subscription request handlers.
const streamOptions = {
subscriptionRequestHandler: subscriptionRequest => subscriptionRequest.accept(),
subscriptionAddedHandler: console.log,
subscriptionRemovedHandler: console.log
};
// Creating the stream.
let stream;
async function initiateStream() {
stream = await io.interop.createStream(streamDefinition, streamOptions);
console.log(`Stream "${stream.definition.displayName}" created successfully.`);
};
initiateStream().catch(console.error);
Accepting or Rejecting Subscriptions
Subscriptions are auto accepted by default. You can control this behavior by passing a subscriptionRequestHandler
in the StreamOptions
object. This handler is called before the subscriptionAddedHandler
, so if you reject the request, the subscriptionAddedHandler
won't be called.
The SubscriptionRequest
object, passed as an argument to the subscription request handler, has the following properties:
Property | Type | Description |
---|---|---|
accept() |
function |
Accepts the instance subscription. |
acceptOnBranch() |
function |
Accepts the subscription on a branch with the provided string argument as a name. Pushing data to that branch will multicast it to all subscriptions associated with the branch. |
arguments |
object |
Object containing the subscription arguments. |
instance |
object |
The Interop Instance of the subscriber app. |
reject() |
function |
Rejects the subscription and returns the provided string argument as a reason for the rejection. |
Example of a subscription request handler:
function onSubscriptionRequest(subscriptionRequest) {
// Here you can identify, accept or reject subscribers,
// group subscribers on a shared stream branch, access the subscription arguments.
const application = subscriptionRequest.instance.application;
const symbol = subscriptionRequest.arguments.symbol;
// If the subscription request contains a `symbol` property in the its `arguments` object,
// accept it on a stream branch with the provided symbol as a branch key,
// otherwise, reject the subscription.
if (symbol) {
subscriptionRequest.acceptOnBranch(symbol);
console.log(`Accepted subscription by "${application}" on branch "${symbol}".`);
} else {
subscriptionRequest.reject("Subscription rejected: missing `symbol` argument.");
console.warn(`Rejected subscription by "${application}". Symbol not specified.`);
};
};
Added & Removed Subscriptions
By default, nothing happens when a new subscription is added or removed. You may, however, want to push data to the subscriber, if the data is available, or unsubscribe from the underlying data source, when the last subscriber for that data is removed. Use the subscriptionAddedHandler
and the subscriptionRemovedHandler
in the StreamOptions
object to achieve this.
Handling New Subscriptions
Example of a handler for added subscriptions:
const symbolPriceCache = {
"GOOG": {
price: 123.456
}
};
function onSubscriptionAdded(streamSubscription) {
const symbol = streamSubscription.arguments.symbol;
const isFirstSubscription = symbolPriceCache[symbol] ? false : true;
if (isFirstSubscription) {
// If this is a first subscription for that symbol,
// start requesting data for it and cache it.
symbolPriceCache[symbol] = {};
startDataRequests(symbol);
console.log(`First subscription for symbol "${symbol}" created.`);
} else {
// If there is already an existing subscription for that symbol,
// send a snapshot of the available price to the new subscriber.
const price = symbolPriceCache[symbol].price;
// Check first whether a price is available.
if (price) {
const data = { symbol, price };
// Unicast data directly to this subscriber.
streamSubscription.push(data);
console.log(`Sent snapshot price for symbol "${symbol}".`);
};
};
};
function startDataRequests(symbol) {
// Here you can make requests to a real-time data source.
};
Handling Last Subscription Removal
Example of a handler for removed subscriptions:
function onSubscriptionRemoved(streamSubscription) {
const symbol = streamSubscription.arguments.symbol;
const branch = streamSubscription.stream.branch(symbol);
// If there are no more subscriptions for that symbol,
// stop requesting data and remove the symbol from the cache.
if (branch === undefined) {
stopDataRequests(symbol);
delete symbolPriceCache[symbol];
console.warn(`Branch was closed, no more active subscriptions for symbol "${symbol}".`);
};
};
function stopDataRequests(symbol) {
// Terminate the requests to the data source.
};
Using Stream Branches
If your stream publishing code uses branches (e.g., creates a branch for each unique set of subscription arguments and associates the subscriptions with that branch), whenever a data arrives from your underlying source, you can use the branch to publish the data to the subscribers on that branch instead of manually going over all subscriptions and pushing data to the interested clients.
Example:
// Extract the data returned in the response from the data source, e.g.:
// const symbol = responseData.symbol;
// const price = responseData.price;
const data = { symbol, price };
// The subscriptions have been accepted on branches with the `symbol`
// provided in the subscription requests as a branch key,
// so now the same `symbol` is used to identify the branch to which to push data.
stream.push(data, symbol);
Server Side Subscription Object
The StreamSubscription object has the following properties:
Property | Type | Description |
---|---|---|
arguments |
object |
The arguments used by the client app to subscribe. |
branchKey |
string |
The key of the branch (if any) with which the stream publisher has associated the client subscription. |
close() |
function |
method which closes the subscription forcefully on the publisher side, e.g. if the publisher shuts down. |
instance |
object |
Instance object of the subscriber. |
push() |
function |
A method to push data directly to a subscription (unicast). |
stream |
object |
The Stream object you have registered. |
Stream Object
The Stream object has the following properties:
Property | Type | Description |
---|---|---|
branches() |
function |
Returns a list of all branches. |
close() |
function |
Closes the stream and unregisters the corresponding Interop method. |
definition |
object |
The MethodDefinition object with which the stream was created. |
name |
string |
The name of the stream as specified in the definition object. |
subscriptions() |
function |
Returns a list of all subscriptions. |
Branch Object
The StreamBranch object has the following properties:
Property | Type | Description |
---|---|---|
close() |
function |
Closes the branch (and drops all subscriptions on it). |
key |
string |
The key with which the branch was created. |
push() |
function |
Multicasts data to all subscriptions on the branch. This is always more efficient than keeping track of the subscriptions manually and doing it yourself. |
subscriptions() |
function |
Returns all subscriptions which are associated with this branch. |
Consuming Stream Data
Subscribing to Streams
Streams are simply special Interop methods, so subscribing to a stream resembles very much invoking a method. To subscribe, create a subscription using the subscribe()
method. It accepts a string or a MethodDefinition
object as a first required argument and a SubscriptionParams
object as a second optional one:
const subscriptionOptions = {
arguments: { symbol: "GOOG" }
};
// Creating the subscription.
let subscription;
async function createSubscription() {
subscription = await io.interop.subscribe("MarketData.LastTrades", subscriptionOptions);
};
createSubscription().catch(console.error);
// Use subscription here.
The SubscriptionParams
object has the following properties:
Property | Type | Description |
---|---|---|
arguments |
object |
Object containing arguments for the stream subscription. Passing arguments enables you to group subscribers that use the same arguments on a stream branch (see Publishing Stream Data), or use these as a filter on the publisher side. |
methodResponseTimeout |
object |
Interval in milliseconds to wait for the stream reply. |
onClosed |
function |
Callback to handle the event when the subscription is closed by the server. |
onConnected |
function |
Callback to handle the event when the subscription is connected to a server. |
onData |
function |
Callback for handling new data. |
target |
string | object |
InstanceTarget enumeration that can be one of "best" , "all" , "skipMine" , Instance or an array of Instance objects (see Method Invocation). |
waitTimeoutMs |
number |
Interval in milliseconds to wait for discovering the stream if not immediately available. |
Handling Subscriptions Client Side
The client side Subscription
object has several useful properties providing information about the subscription instance:
Property | Type | Description |
---|---|---|
requestArguments |
object |
Arguments used for the subscription. |
servers |
object[] |
Array of Instance objects of the apps providing the stream. |
serverInstance |
object |
Instance object of the app providing the stream. |
stream |
object |
The stream definition object. |
Once you have a subscription, use its onData()
method to handle stream data. The callback you register with the onData()
method of the Subscription
object will fire every time new stream data is received:
subscription.onData((streamData) => {
// Use stream data here.
});
The StreamData
object has the following properties:
Property | Type | Description |
---|---|---|
data |
object |
The data object sent by the stream publisher. |
message |
string |
Message from the publisher of the stream. |
private |
boolean |
Flag indicating whether the data was unicast to this subscription (false , if multicast from a stream or a stream branch). |
requestArguments |
object |
The subscription request arguments. |
server |
object |
Instance object of the Interop server which pushed the data. |
Closed or Rejected Subscriptions
A stream subscription can be closed at any time due to the publisher shutting down or due to an error. Two methods handle these events:
subscription.onClosed(() => {
// Closed gracefully by the publisher.
});
subscription.onFailed((error) => {
// Unexpected error in the publisher.
});
Stream Discovery
Streams are special Interop methods, so you can use the Interop discovery methods to find available streams. The only difference is that streaming methods are flagged with a property supportsStreaming: true
.
Finding all streams:
const streams = io.interop.methods().filter(method => method.supportsStreaming === true);
Finding a known stream:
const stream = io.interop.methods().find(method => method.name === "MarketData.LastTrades");
Complete Streaming Example
Below is a complete streaming example containing server side and client side code. The server publishes a stream that simulates fetching real time market data for financial instruments from a data source and sending it to subscribers by using stream branches. A new branch is created for each unique instrument symbol
(the branch key
is the symbol
itself). The client subscribes for this data by providing the symbol
of the financial instrument in the subscription request. The server defines callbacks for handling the subscription, and the client defines callbacks for handling new data and the event which fires when the server closes the subscription. Logging is provided both on the server and on the client side to allow you to follow the streaming events more easily.
To test the example:
Open the console of an interop-enabled app and paste the Stream Publisher example in it.
Open the console of a different interop-enabled app and paste the Stream Consumer example in it.
To simulate multiple subscriptions, repeat step 2 with several other interop-enabled apps. Change the value of the
symbol
property (use any random string) in thearguments
object of the subscription options to simulate subscriptions for different financial instruments. (Don't forget to change the names of the variables that have already been declared if you want to make more than one subscription from the same client.)Experiment with the
stream
,stream.branches()
andsubscription
objects in the console to trigger the event handlers. For more information on what you can do, explore the Interop API reference documentation.
Stream Publisher
// Cache object that will contain all symbols and symbol prices
// for which there are active subscriptions.
const symbolPriceCache = {};
// Variable that will hold the stream object.
let stream;
/** SUBSCRIPTION HANDLERS **/
function onSubscriptionRequest(subscriptionRequest) {
const application = subscriptionRequest.instance.application;
const symbol = subscriptionRequest.arguments.symbol;
// If the subscription request contains a `symbol` property in the its `arguments` object,
// accept it on a stream branch with the provided symbol as a branch key,
// otherwise, reject the subscription.
if (symbol) {
subscriptionRequest.acceptOnBranch(symbol);
console.log(`Accepted subscription by "${application}" on branch "${symbol}".`);
} else {
subscriptionRequest.reject("Subscription rejected: missing `symbol` argument.");
console.warn(`Rejected subscription by "${application}". Symbol not specified.`);
};
};
function onSubscriptionAdded(streamSubscription) {
const symbol = streamSubscription.arguments.symbol;
const isFirstSubscription = symbolPriceCache[symbol] ? false : true;
if (isFirstSubscription) {
// If this is a first subscription for that symbol,
// start requesting data for it and cache it.
symbolPriceCache[symbol] = {};
startDataRequests(symbol);
console.log(`First subscription for symbol "${symbol}" created.`);
} else {
// If there is already an existing subscription for that symbol,
// send a snapshot of the available price to the new subscriber.
const price = symbolPriceCache[symbol].price;
// First check first whether a price is available.
if (price) {
const data = { symbol, price };
streamSubscription.push(data);
console.log(`Sent snapshot price for symbol "${symbol}".`);
};
};
};
function onSubscriptionRemoved(streamSubscription) {
const symbol = streamSubscription.arguments.symbol;
const branch = streamSubscription.stream.branch(symbol);
// If there are no more subscriptions for that symbol,
// stop requesting data and remove the symbol from the cache.
if (branch === undefined) {
stopDataRequests(symbol);
delete symbolPriceCache[symbol];
console.warn(`Branch was closed, no more active subscriptions for symbol "${symbol}".`);
};
};
/** PUBLISHING THE STREAM **/
// Stream definition.
const streamDefinition = {
name: "MarketData.LastTrades",
displayName: "Market Data - Last Trades",
accepts: "String symbol",
returns: "String symbol, Double lastTradePrice"
};
// Stream options object containing subscription request handlers.
const streamOptions = {
subscriptionRequestHandler: onSubscriptionRequest,
subscriptionAddedHandler: onSubscriptionAdded,
subscriptionRemovedHandler: onSubscriptionRemoved
};
// Creating the stream.
async function initiateStream() {
stream = await io.interop.createStream(streamDefinition, streamOptions);
console.log(`Stream "${stream.definition.displayName}" created successfully.`);
};
initiateStream().catch(console.error);
/** HELPER FUNCTIONS **/
function startDataRequests(symbol) {
// Set up a task to send requests to a data source every 5 seconds.
symbolPriceCache[symbol].pollingTask = setInterval(fetchMarketData, 5000, symbol);
};
function stopDataRequests(symbol) {
const pollingTask = symbolPriceCache[symbol].pollingTask;
// Stop the requests to the data source.
clearInterval(pollingTask);
};
function fetchMarketData(symbol) {
// Here is the place to create actual requests to a data source
// and push the received data to the subscribers on the respective branch.
const price = Math.random() * 1000;
const data = { symbol, price };
// Push the `data` to all subscribers on the branch with key `symbol`.
stream.push(data, symbol);
// Cache the price for the symbol in order to use it
// as a snapshot for a subsequent subscription.
symbolPriceCache[symbol].price = price;
};
Stream Consumer
// Subscription options object containing subscription arguments
// and callbacks to handle new data and closing the subscription.
const subscriptionOptions = {
arguments: { symbol: "GOOG" },
onData: (streamData) => console.log(streamData.data.symbol, streamData.data.price),
onClosed: () => console.warn("Subscription closed by server.")
};
// Creating the subscription.
let subscription;
async function createSubscription() {
subscription = await io.interop.subscribe("MarketData.LastTrades", subscriptionOptions);
};
createSubscription().catch(console.error);
Reference
For a complete list of the available Interop API methods and properties, see the Interop API Reference Documentation.