An Overview of Diffusion Intelligent Event Data Platform and Its Capabilities

90%

Over 90% reduction in bandwidth used

1M Connections

Supports over 1M concurrent connections with a real-time event broker

1M Topics

Supports over 1M topics using a hierarchical data model

Bring new capabilities to your applications

Gaining and keeping customers can be highly dependent on application responsiveness. We understand creating and evolving applications can be costly, so we made adding real-time and personalization capabilities easier.

Data Gateway
Data Wrangling
Data Distribution

DATA GATEWAY

Makes it easy to integrate with any data source within your data center or outside with pre-built Adapters for Kafka, MQTT, REST, JMS or backend applications using our WebSocket based SDKs in all popular languages, Accommodates both streaming and static data sources, Supports any size, format, or velocity of data.

What We Provide Our Customers

Faster time-to-market and efficient operation in production

BUILD
PERFORMANCE
COST
CONTROL
SECURE

A holistic platform with low-code features to simplify coding efforts for faster time-to-market

WebSocket based SDKs in all popular programming languages
Low-Code Features to simplify coding efforts to organize data
Easy to integrate with any data sources with pre-built adapters
Choice of Communication methods – Pub-Sub, Req-Resp, Event History
REST API, Open-Protocols and Adapters to integrate legacy Apps.

Bring new possibilities to your application

1. Real-Time Event Broker
2. Delta Streaming
3. Hierarchical Topic Trees
4. Low Code Features
5. Open Protocols
6. Fine-Grained Security
7. Seamless Scalability
8. Analytics Dashboard
9. Mesh Architecture
10. Integrations

PUB-SUB

One of four delivery mechanisms available inside Diffusion enabling you to publish real-time data to hundreds of thousands of consumers efficiently.

REQUEST-RESPONSE

Provide reliable two-way messaging between multiple endpoints allowing you to build sophisticated asynchronous interactions

TIME SERIES (EVENT HISTORY)

Store real-time events and dynamically query historical data enabling you to replay past events at any point in time.

FETCH

Retrieve snapshots from real-time streams and efficiently access static data through a simple query API.

How our Product Works

An overview of Diffusion Intelligent Event Data Platform and its capabilities

Our Customers

Diffusion is a secure, robust, fast & flexible real-time platform which supports our Electronic Trading Platform & the features within our GUI.

Yann L’Huillier

Group CIO, Tradition

Founded in 1959, Tradition is one of the world leaders in the IDB (Interdealer broker) sector of financial and non-financial products.

PUSH Diffusion was selected to deliver real-time pricing data, research reports, trade alerts and risk reports for Tradition’s new cloud-based data and & analytics SaaS offerings. Diffusion scales to meet the trade volumes Tradition needs, with lower compute and network capacity demands. Learn more – here.

Our Customers

A huge benefit of Diffusion is the efficient distribution of data to support real-time mobile apps and allow us to provide a seamless customer experience.

Ed Braidwood

IT Operations Manager

Racing Post is the online home of the UK’s number one horse racing daily newspaper. Its website, bookkeeper kiosks, and mobile apps have become the online destination of choice for horse racing enthusiasts.

With Diffusion, in addition to providing instant results, new value-added services such as in-race, live-text commentaries, analytical statistics, race status data, and lets customers know when new website content – such as a live news ticker – is available. Learn more - here.

Our Customers

We have set-up a completely cloud-based infrastructure and we can develop, test, and go into production with Diffusion Cloud for free – a huge benefit for a young company.

Philipp Roeben

Co-Founder & CIO

Founded in 2018 and headquartered in Munich, GRATISBROKER, offers free trading of more than 3,600 stocks, around 180 ETFs and over 2,100 funds.

To provide, live, event-driven information in real-time to many thousands of customers online, the GRATISBROKER platform is built on Diffusion. Now, the trading platform provides high quality-of-service along with the scalability to meet demand peaks. Learn more - here.

Simplicity in Action

Our powerful real-time SDKs and Service API make building applications simple. See how easy it is to create streaming and REST based applications.

Publish-Subscribe
REST
Request-Response
Time Series

JavaScript
.NET
Android
Apple
Java
C
Python

Publish

// Add a topic and set its value
diffusion.connect({
    host : "",
    port : "",
    principal : "admin",
    credentials : ""
}).then(function(session) {
    // Use the new session here
});

session.topics.add('my-topic', diffusion.topics.TopicType.JSON);
var value = { foo : "bar" };
var result = session.topicUpdate.set("my-topic", diffusion.datatypes.json(), value);

// Subscribe to topics.
diffusion.connect({
    host : "",
    port : "",
    principal : "admin",
    credentials : ""
}).then(function(session) {
    // Use the new session here
});

session.addStream('my-topic',
                  diffusion.datatypes.json()).on('value',
                  function(topic, specification, newValue, oldValue) {
  console.log('New value:', newValue.get());
});

session.select('my-topic');

Publish

// Add a topic and set its value.
var session:PTDiffusionSession?

let url = URL(string: "ws://:")!
let credentials = PTDiffusionCredentials(password: "")
let config = PTDiffusionSessionConfiguration(principal: "admin", credentials: credentials)

PTDiffusionSession.open(with: url, configuration: config) { (_session, error) -> Void in
    self.session = _session
}

let errorHandler: (Any?, Error?) -> Void = {response, error in
    if (error != nil) {
        print(error!)
    }
}

self.session!.topicControl.addTopic(withPath: "my-topic",
                                    type: PTDiffusionTopicType.JSON,
                                    completionHandler: self.errorHandler)

let value = try? PTDiffusionJSON(object: ["foo": "bar"])

self.session!.topicUpdate.setWithPath("my-topic",
                                      toJSONValue: value!) { error in self.errorHandler(nil, error) }

// Subscribe to topics.
class StreamDelegate: NSObject, PTDiffusionJSONValueStreamDelegate {

    func diffusionStream(_ stream: PTDiffusionStream,
                         didUnsubscribeFromTopicPath topicPath: String,
                         specification: PTDiffusionTopicSpecification,
                         reason: PTDiffusionTopicUnsubscriptionReason) {
        print("Unsubscribed from: \(topicPath)")
    }

    func diffusionStream(_ stream: PTDiffusionStream,
                         didFailWithError error: Error) {
        print("Failed with error: \(error)")
    }

    func diffusionDidClose(_ stream: PTDiffusionStream) {
        print("Closed")
    }

    func diffusionStream(_ stream: PTDiffusionStream,
                         didSubscribeToTopicPath topicPath: String,
                         specification: PTDiffusionTopicSpecification) {
        print("Subscribed to: \(topicPath)")
    }

    func diffusionStream(_ stream: PTDiffusionValueStream,
                         didUpdateTopicPath topicPath: String,
                         specification: PTDiffusionTopicSpecification,
                         oldJSON: PTDiffusionJSON?,
                         newJSON: PTDiffusionJSON) {
        do {
            let value:Dictionary = try newJSON.object() as! Dictionary
            print("\(topicPath): \(value.description)")
        }
        catch {
            print("Unable to read message")
        }
    }
}

var session:PTDiffusionSession?

let url = URL(string: "ws://:")!
let credentials = PTDiffusionCredentials(password: "")
let config = PTDiffusionSessionConfiguration(principal: "admin", credentials: credentials)

PTDiffusionSession.open(with: url, configuration: config) { (_session, error) -> Void in
    self.session = _session
}

let errorHandler: (Any?, Error?) -> Void = {response, error in
    if (error != nil) {
        print(error!)
    }
}

let delegate = StreamDelegate()

let selector = PTDiffusionTopicSelector(expression: "my-topic")
let stream = PTDiffusionJSON.valueStream(with: self.delegate)

try? self.session!.topics.add(stream, with: selector, error: ())

self.session!.topics.subscribe(withTopicSelectorExpression: "my-topic") { error in self.errorHandler(nil, error) }

Publish

/*
 *  Add a topic and set its value.
 */
DIFFUSION_SESSION_FACTORY_T *session_factory = diffusion_session_factory_init();
diffusion_session_factory_principal(session_factory, "admin");
diffusion_session_factory_password(session_factory, "");

SESSION_T *session = session_create_with_session_factory(session_factory, "ws://::");

on_topic_added(SESSION_T *session, TOPIC_ADD_RESULT_CODE result_code, void *context)
{
        printf("on_topic_added\n");
        return HANDLER_SUCCESS;
}

static int
on_topic_add_failed(SESSION_T *session, TOPIC_ADD_FAIL_RESULT_CODE result_code, const DIFFUSION_ERROR_T *error, void *context)
{
        printf("on_topic_add_failed: %d\n", result_code);
        return HANDLER_SUCCESS;
}

static int
on_topic_update(void *context)
{
        printf("topic update success\n");
        return HANDLER_SUCCESS;
}

ADD_TOPIC_CALLBACK_T callback = {
        .on_topic_added_with_specification = on_topic_added,
        .on_topic_add_failed_with_specification = on_topic_add_failed
};

TOPIC_SPECIFICATION_T *spec = topic_specification_init(TOPIC_TYPE_JSON);
add_topic_from_specification(session, "my-topic", spec, callback);

BUF_T *buf = buf_create();
write_diffusion_json_value("\"hello world\"", buf);

DIFFUSION_TOPIC_UPDATE_SET_PARAMS_T topic_update_params = {
        .topic_path = "my-topic",
        .datatype = DATATYPE_JSON,
        .update = buf,
        .on_topic_update = on_topic_update
};

diffusion_topic_update_set(session, topic_update_params);

/*
 *  Subscribe to topics.
 */
DIFFUSION_SESSION_FACTORY_T *session_factory = diffusion_session_factory_init();
diffusion_session_factory_principal(session_factory, "admin");
diffusion_session_factory_password(session_factory, "");

SESSION_T *session = session_create_with_session_factory(session_factory, "ws://::");


static int on_subscription(const char* topic_path,
                    const TOPIC_SPECIFICATION_T *specification,
                    void *context)
{
        printf("Subscribed to topic: %s\n", topic_path);
        return HANDLER_SUCCESS;
}

static int on_unsubscription(const char* topic_path,
                      const TOPIC_SPECIFICATION_T *specification,
                      NOTIFY_UNSUBSCRIPTION_REASON_T reason,
                      void *context)
{
        printf("Unsubscribed from topic: %s\n", topic_path);
        return HANDLER_SUCCESS;
}

static int on_value(const char* topic_path,
             const TOPIC_SPECIFICATION_T *const specification,
             const DIFFUSION_DATATYPE datatype,
             const DIFFUSION_VALUE_T *const old_value,
             const DIFFUSION_VALUE_T *const new_value,
             void *context)
{
        char *result;
        bool success = to_diffusion_json_string(new_value, &result, &api_error);

        if(success) {
                printf("Received value: %s\n", result);
                free(result);
                return HANDLER_SUCCESS;
        }

        DIFFUSION_API_ERROR api_error;
        printf("Error during diffusion value read: %s\n", get_diffusion_api_error_description(api_error));
        diffusion_api_error_free(api_error);
        return HANDLER_FAILURE;
}

static void on_close()
{
        printf("Value stream closed\n");
}

VALUE_STREAM_T value_stream = {
        .datatype = DATATYPE_JSON,
        .on_subscription = on_subscription,
        .on_unsubscription = on_unsubscription,
        .on_value = on_value,
        .on_close = on_close
};

add_stream(session, "my-topic", &value_stream);
SUBSCRIPTION_PARAMS_T params = {
        .topic_selector = "my-topic"
};

subscribe(session, params);

JavaScript
.NET
Android
Apple
Java
C
Python

Publish

//Create a topic with the REST API.
// should be replaced with the full access token
//obtained Diffusion Cloud service dashboard.
fetch("https://api.diffusion.cloud/topics/add", {
  "method": "POST",
  "headers": {
    "authorization:": "Bearer ",
    "content-type": "application/json"
  },
  "body": {
    "path": "my-topic",
    "type": "json",
    "properties": {
      "VALIDATE_VALUES": "true"
    }
  }
})
.then(response => {
  console.log(response);
})
.catch(err => {
  console.log(err);
});

//Update the topic with a value.
fetch("https://api.diffusion.cloud/topics/set", {
  "method": "POST",
  "headers": {
    "authorization:": "Bearer ",
    "content-type": "application/json"
  },
  "body": {
    "path": "my-topic",
    "type": "json",
    "value": {
      "foo": "bar"
    }
  }
})
.then(response => {
  console.log(response);
})
.catch(err => {
  console.log(err);
});

//Get the current value of the topic.
// should be replaced with the full access token
//obtained Diffusion Cloud service dashboard.
fetch("https://api.diffusion.cloud/topics/fetch?path=my-topic", {
  "method": "GET",
  "headers": {
    "authorization:": "Bearer "
  }
})
.then(response => {
  console.log(response);
})
.catch(err => {
  console.log(err);
});

Publish

//Create a topic with the REST API.
// should be replaced with the full access token
//obtained Diffusion Cloud service dashboard.
URL url = new URL ("https://api.diffusion.cloud/topics/add");
HttpURLConnection connection = (HttpURLConnection)url.openConnection();
connection.setRequestMethod("POST");
connection.setRequestProperty("Authorization", "Bearer ");
connection.setRequestProperty("Content-Type", "application/json; utf-8");
connection.setRequestProperty("Accept", "application/json");
connection.setDoOutput(true);
String requestParameter = "{\"path\":\"my-topic\", \"type\":\"json\", \"properties\":{ \"VALIDATE_VALUES\":\"true\" }}";
try(OutputStream os = connection.getOutputStream()) {
    byte[] input = requestParameter.getBytes("utf-8");
    os.write(input, 0, input.length);
}
InputStream is = connection.getInputStream();

//Update the topic with a value.
URL urlUpdate = new URL ("https://api.diffusion.cloud/topics/set");
HttpURLConnection connectionUpdate = (HttpURLConnection)urlUpdate.openConnection();
connectionUpdate.setRequestMethod("POST");
connectionUpdate.setRequestProperty("Authorization", "Bearer ");
connectionUpdate.setRequestProperty("Content-Type", "application/json; utf-8");
connectionUpdate.setRequestProperty("Accept", "application/json");
connectionUpdate.setDoOutput(true);
String requestParameterUpdate = "{\"path\":\"my-topic\", \"type\":\"json\", \"value\": {\"foo\": \"bar\"}}";
try(OutputStream osUpdate = connectionUpdate.getOutputStream()) {
    byte[] input = requestParameterUpdate.getBytes("utf-8");
    osUpdate.write(input, 0, input.length);
}
InputStream isUpdate = connectionUpdate.getInputStream();

//Get the current value of the topic.
// should be replaced with the full access token
//obtained Diffusion Cloud service dashboard.
URL url = new URL("https://api.diffusion.cloud/topics/fetch?path=my-topic");
HttpURLConnection connection = (HttpURLConnection)url.openConnection();

connection.setRequestMethod("GET");
connection.setRequestProperty("Authorization", "Bearer ");

try(InputStream inputStream = connection.getInputStream()) {
  InputStreamReader inputStreamReader = new InputStreamReader(inputStream);
  BufferedReader bufferedReader = new BufferedReader(inputStreamReader);
  
  String inputLine;
  StringBuffer response = new StringBuffer();

  while ((inputLine = bufferedReader.readLine()) != null) {
    response.append(inputLine);
  }
}

Publish

//Create a topic with the REST API.
// should be replaced with the full access token
//obtained Diffusion Cloud service dashboard.
import Foundation

let headers = [
  "authorization:": "Bearer ",
  "content-type": "application/json"
]
let parameters = [
  "path": "my-topic",
  "type": "json",
  "properties": ["VALIDATE_VALUES": "true"]
] as [String : Any]

let postData = JSONSerialization.data(withJSONObject: parameters, options: [])

let request = NSMutableURLRequest(url: NSURL(string: "https://api.diffusion.cloud/topics/add")! as URL,
                                        cachePolicy: .useProtocolCachePolicy,
                                    timeoutInterval: 10.0)
request.httpMethod = "POST"
request.allHTTPHeaderFields = headers
request.httpBody = postData as Data

let session = URLSession.shared
let dataTask = session.dataTask(with: request as URLRequest, completionHandler: { (data, response, error) -> Void in
  if (error != nil) {
    print(error)
  } else {
    let httpResponse = response as? HTTPURLResponse
    print(httpResponse)
  }
})

dataTask.resume()

//Update the topic with a value.
let parametersUpdate = [
  "path": "my-topic",
  "type": "json",
  "value": ["foo": "bar"]
] as [String : Any]

let postDataUpdate = JSONSerialization.data(withJSONObject: parametersUpdate, options: [])

let requestUpdate = NSMutableURLRequest(url: NSURL(string: "https://api.diffusion.cloud/topics/set")! as URL,
                                        cachePolicy: .useProtocolCachePolicy,
                                    timeoutInterval: 10.0)
requestUpdate.httpMethod = "POST"
requestUpdate.allHTTPHeaderFields = headers
requestUpdate.httpBody = postDataUpdate as Data

let sessionUpdate = URLSession.shared
let dataTaskUpdate = sessionUpdate.dataTask(with: requestUpdate as URLRequest, completionHandler: { (data, response, error) -> Void in
  if (error != nil) {
    print(error)
  } else {
    let httpResponseUpdate = response as? HTTPURLResponse
    print(httpResponseUpdate)
  }
})

dataTaskUpdate.resume()

//Get the current value of the topic.
// should be replaced with the full access token
//obtained Diffusion Cloud service dashboard.
import Foundation

let headers = ["authorization:": "Bearer "]

let request = NSMutableURLRequest(url: NSURL(string: "https://api.diffusion.cloud/topics/fetch?path=my-topic")! as URL,
                                        cachePolicy: .useProtocolCachePolicy,
                                    timeoutInterval: 10.0)
request.httpMethod = "GET"
request.allHTTPHeaderFields = headers

let session = URLSession.shared
let dataTask = session.dataTask(with: request as URLRequest, completionHandler: { (data, response, error) -> Void in
  if (error != nil) {
    print(error)
  } else {
    let httpResponse = response as? HTTPURLResponse
    print(httpResponse)
  }
})

dataTask.resume()

Publish

//Create a topic with the REST API.
// should be replaced with the full access token
//obtained Diffusion Cloud service dashboard.
URL url = new URL ("https://api.diffusion.cloud/topics/add");
HttpURLConnection connection = (HttpURLConnection)url.openConnection();
connection.setRequestMethod("POST");
connection.setRequestProperty("Authorization", "Bearer ");
connection.setRequestProperty("Content-Type", "application/json; utf-8");
connection.setRequestProperty("Accept", "application/json");
connection.setDoOutput(true);
String requestParameter = "{\"path\":\"my-topic\", \"type\":\"json\", \"properties\":{ \"VALIDATE_VALUES\":\"true\" }}";
try(OutputStream os = connection.getOutputStream()) {
    byte[] input = requestParameter.getBytes("utf-8");
    os.write(input, 0, input.length);
}
InputStream is = connection.getInputStream();

//Update the topic with a value.
URL urlUpdate = new URL ("https://api.diffusion.cloud/topics/set");
HttpURLConnection connectionUpdate = (HttpURLConnection)urlUpdate.openConnection();
connectionUpdate.setRequestMethod("POST");
connectionUpdate.setRequestProperty("Authorization", "Bearer ");
connectionUpdate.setRequestProperty("Content-Type", "application/json; utf-8");
connectionUpdate.setRequestProperty("Accept", "application/json");
connectionUpdate.setDoOutput(true);
String requestParameterUpdate = "{\"path\":\"my-topic\", \"type\":\"json\", \"value\": {\"foo\": \"bar\"}}";
try(OutputStream osUpdate = connectionUpdate.getOutputStream()) {
    byte[] input = requestParameterUpdate.getBytes("utf-8");
    osUpdate.write(input, 0, input.length);
}
InputStream isUpdate = connectionUpdate.getInputStream();

//Get the current value of the topic.
// should be replaced with the full access token
//obtained Diffusion Cloud service dashboard.
URL url = new URL("https://api.diffusion.cloud/topics/fetch?path=my-topic");
HttpURLConnection connection = (HttpURLConnection)url.openConnection();

connection.setRequestMethod("GET");
connection.setRequestProperty("Authorization", "Bearer ");

try(InputStream inputStream = connection.getInputStream()) {
  InputStreamReader inputStreamReader = new InputStreamReader(inputStream);
  BufferedReader bufferedReader = new BufferedReader(inputStreamReader);
  
  String inputLine;
  StringBuffer response = new StringBuffer();

  while ((inputLine = bufferedReader.readLine()) != null) {
    response.append(inputLine);
  }
}

Javascript
.NET
Android
Apple
Java
C
Python

Publish

//Send messages.
var jsonType = diffusion.datatypes.json();
// Create a JSON object to send as a request.
var requestJson = jsonType.from("Hello");

// Send the request to a message path "path".
session.messages.sendRequest('path', requestJson, jsonType).then(function(response) {
console.log(response.get());
}, function(error) {});

//Receive messages.
var jsonType = diffusion.datatypes.json();

var responseJson = jsonType.from({ "ying": "yang"});

// Define a request handler for json topic type
var handler = {
    onRequest: function(request, context, responder) {
        responder.respond(responseJson, jsonType);
    },
    onError: function() {},
    onClose: function() {}
};

session.messages.addRequestHandler('path', handler);

Publish

/*
 * Send messages.
 */
char* request_path = ">random/requestResponse";

SESSION_T *session = session_create("wss://:", "admin", "", NULL, NULL, &error);

BUF_T *request = buf_create();

write_diffusion_string_value("Hello", request);

SEND_REQUEST_PARAMS_T send_request_params = {
        .path = request_path,
        .request = request,
        .on_response = on_response,
        .request_datatype = DATATYPE_STRING,
        .response_datatype = DATATYPE_STRING
};

int counter = 1;

while (counter <= 120) {
        printf("Sending request to path {%s}.. #%d\n", request_path, counter);
        send_request(session, send_request_params);
        sleep(1);
        ++counter;
}

/*
 * Receive messages.
 */
char* request_path = ">random/requestResponse";

SESSION_T *handler = session_create("wss://:", "admin", "", NULL, NULL, &error);

DIFFUSION_REQUEST_HANDLER_T request_handler = {
        .request_datatype = DATATYPE_STRING,
        .response_datatype = DATATYPE_STRING,
        .on_request = on_request
};

ADD_REQUEST_HANDLER_PARAMS_T request_handler_params = {
        .path = request_path,
        .request_handler = &request_handler
};

add_request_handler(handler, request_handler_params);


static int
on_request(SESSION_T *session,  DIFFUSION_DATATYPE request_datatype, const DIFFUSION_VALUE_T *request,
           const DIFFUSION_REQUEST_CONTEXT_T *request_context, const DIFFUSION_RESPONDER_HANDLE_T *handle, void *context)
{
        char *request_val;
        read_diffusion_string_value(request, &request_val, NULL);

        SESSION_ID_T *session_id = diffusion_request_context_get_session_id(request_context);
        char *session_id_str = session_id_to_string(session_id);

        printf("Request received from %s: %s\n", session_id_str, request_val);
        free(request_val);

        BUF_T *response_buf = buf_create();
        write_diffusion_string_value(response, response_buf);
        diffusion_respond_to_request(session, handle, response_buf, NULL);

        buf_free(response_buf);
        session_id_free(session_id);
        free(session_id_str);

        return HANDLER_SUCCESS;
}

Javascript
.NET
Android
Apple
Java
C

Publish

// Add a time series topic and set its value.
var diffusion = require('diffusion');

var TopicSpecification = diffusion.topics.TopicSpecification;
var TopicType = diffusion.topics.TopicType;
var dataType = diffusion.datatypes.double();

diffusion.connect({
    host : "",
    port : "",
    principal : "admin",
    credentials : ""
}).then(function(session) {
    var specification = new TopicSpecification(TopicType.TIME_SERIES, {
        TIME_SERIES_EVENT_VALUE_TYPE : "double"
    });

    session.topics.add('topic/timeseries', specification).then(function() {
        for (var i = 0; i < 10; i++) {
            session.timeseries.append("topic/timeseries", i, dataType);
        }
    });
});

// Subscribe to time series topics.
var diffusion = require('diffusion');

var dataType = diffusion.datatypes.double();

diffusion.connect({
    host : "",
    port : "",
    principal : "admin",
    credentials : ""
}).then(function(session) {
    session.addStream('topic/timeseries', dataType).on('value', function(topic, specification, newValue, oldValue) {
        var value = newValue.toString();

        if (newValue.isEditEvent === true) {
            console.log("New value edited on topic: " + value);
        }
        else {
            console.log("New value appended to topic: " + value);
        }
    });

    session.select('topic/timeseries');    
});

Publish

// Add a time series topic and set its value.
var session = Diffusion.Sessions
    .Principal("admin")
    .Password("")
    .Open("ws://::");

string TOPIC_PREFIX = "time-series";

var typeName = Diffusion.DataTypes.Get().TypeName;
var topic = $"{TOPIC_PREFIX}/{typeName}/{DateTime.Now.ToFileTimeUtc()}";
var specification = session.TopicControl.NewSpecification(TopicType.TIME_SERIES)
    .WithProperty(TopicSpecificationProperty.TimeSeriesEventValueType, typeName);

await session.TopicControl.AddTopicAsync(topic, specification);

// Append a value to the time series topic using a custom timestamp.
await session.TimeSeries.AppendAsync(topic, "Value1", 
                    DateTimeOffset.FromUnixTimeMilliseconds(322));

// Append a value to the time series topic. The timestamp will be set to the current server time.
await session.TimeSeries.AppendAsync(topic, "Value 1");
await session.TimeSeries.AppendAsync(topic, "Value 2");
await session.TimeSeries.AppendAsync(topic, "Value 3");
await session.TimeSeries.AppendAsync(topic, "Value 4");

// Subscribe to time series topics.
string TOPIC_PREFIX = "time-series";

var session = Diffusion.Sessions
    .Principal("admin")
    .Password("")
    .Open("ws://::");

var typeName = Diffusion.DataTypes.Get().TypeName;
var topic = $"?{TOPIC_PREFIX}/{typeName}//";

// Add a value stream
var stringStream = new StringStream();
session.Topics.AddTimeSeriesStream(topic, stringStream);

await session.Topics.SubscribeAsync(topic);


// Basic implementation of the IValueStream for time series string topics.
private sealed class StringStream : IValueStream> {
    public void OnClose() => WriteLine( "The subscription stream is now closed." );

    public void OnError( ErrorReason errorReason )
        => WriteLine( $"An error has occured : {errorReason}." );

    public void OnSubscription( string topicPath, ITopicSpecification specification )
        => WriteLine( $"Client subscribed to {topicPath}." );

    public void OnUnsubscription( string topicPath, ITopicSpecification specification, 
                                    TopicUnsubscribeReason reason )
        => WriteLine( $"Client unsubscribed from {topicPath} : {reason}." );

    public void OnValue( string topicPath, ITopicSpecification specification, 
                            IEvent oldValue, IEvent newValue )
        => WriteLine( $"New value of {topicPath} is {newValue}." );
}

Publish

// Add a time series topic and set its value.
var session:PTDiffusionSession?

let url = URL(string: "ws://:")!
let credentials = PTDiffusionCredentials(password: "")
let config = PTDiffusionSessionConfiguration(principal: "admin", credentials: credentials)

PTDiffusionSession.open(with: url, configuration: config) { (_session, error) -> Void in
    self.session = _session
}

let errorHandler: (Any?, Error?) -> Void = {response, error in
    if (error != nil) {
        print(error!)
    }
}

NSString *const topicPath = @"timeSeries/example";

[self createTimeSeriesTopic:topicPath fromSession:session];

// append value and user supplied timestamp to the topic
NSDateFormatter *dateFormatter = [[NSDateFormatter alloc] init];
[dateFormatter setDateFormat:@"yyyy-MM-dd hh:mm:ss Z"];

NSDate *date1 = [dateFormatter dateFromString:@"1452-04-15 12:34:56 +0000"];
[self appendValue:@"hello world!" withDate:date1 toTimeSeriesTopic:topicPath usingSession:session];


-(void) createTimeSeriesTopic:(NSString *const)topicPath fromSession:(PTDiffusionSession *const)session {

    NSDictionary *properties =
        @{PTDiffusionTopicSpecification.timeSeriesEventValueTypePropertyKey: PTDiffusionDataTypes.json.typeName};

    PTDiffusionTopicSpecification *const specification =
        [[PTDiffusionTopicSpecification alloc] initWithType:PTDiffusionTopicType_TimeSeries
                                                 properties:properties];

    [session.topicControl addTopicWithPath:topicPath
                             specification:specification
                         completionHandler:^(PTDiffusionAddTopicResult * _Nullable result, NSError * _Nullable error)
     {
         if ([result isEqual:PTDiffusionAddTopicResult.exists]) {
             NSLog(@"Topic already existed");
         }
         else if ([result isEqual:PTDiffusionAddTopicResult.created]) {
             NSLog(@"Topic created");
         }
         else if (error != nil) {
             NSLog(@"An error occurred while attempting to create topic: [%@]", error);
             return;
         }
     }];
}

-(void) appendValue:(NSString *const)value
                 withDate:(NSDate *)date
                 toTimeSeriesTopic:(NSString *const)topicPath
                 usingSession:(PTDiffusionSession *const)session {

    NSError *error;
    PTDiffusionJSON *const jsonValue =
        [[PTDiffusionJSON alloc] initWithJSONString:[NSString stringWithFormat:@"{\"value\" : \"%@\"}", value]
                                              error:&error];

    [session.timeSeries appendToTopicPath:topicPath
                                JSONValue:jsonValue
                                timestamp:date
                        completionHandler:^(PTDiffusionTimeSeriesEventMetadata * _Nullable eventMetadata, NSError * _Nullable error)
     {
         if (error != nil) {
             NSLog(@"An error occurred while appending to the topic: [%@]", error);
             return;
         }

         NSString *const author = eventMetadata.author;
         NSDate *const timestamp = [NSDate dateWithTimeIntervalSince1970:eventMetadata.timestamp / 1000];
         NSDateFormatter *const dateFormatter = [[NSDateFormatter alloc] init];
         [dateFormatter setDateFormat:@"yyyy-MM-dd HH:mm:ss"];
         NSLog(@"Time series topic with user supplied timestamp appended by [%@] at [%@]", author, [dateFormatter stringFromDate:timestamp]);
     }];
}

// Subscribe to topics.
var session:PTDiffusionSession?

let url = URL(string: "ws://:")!
let credentials = PTDiffusionCredentials(password: "")
let config = PTDiffusionSessionConfiguration(principal: "admin", credentials: credentials)

PTDiffusionSession.open(with: url, configuration: config) { (_session, error) -> Void in
    self.session = _session
}

let errorHandler: (Any?, Error?) -> Void = {response, error in
    if (error != nil) {
        print(error!)
    }
}

NSString *const topicPath = @"timeSeries/example";
let selector = PTDiffusionTopicSelector(expression: topicPath)

let streamDelegate = [JSONTimeSeriesEventValueStreamDelegate new];
let valueStream = [PTDiffusionJSON timeSeriesEventValueStreamWithDelegate:streamDelegate];

try? self.session!.topics.add(valueStream, with: selector, error: ())

self.session!.topics.subscribe(withTopicSelectorExpression: topicPath) { error in self.errorHandler(nil, error) }

Publish

// Add a time series topic and set its value.
DIFFUSION_SESSION_FACTORY_T *session_factory = diffusion_session_factory_init();
diffusion_session_factory_principal(session_factory, "admin");
diffusion_session_factory_password(session_factory, "");

SESSION_T *session = session_create_with_session_factory(session_factory, "ws://::");

char *topic_path = "time-series-topic";

ADD_TOPIC_CALLBACK_T callback = create_topic_callback(topic_path);

HASH_T *properties = hash_new(2);
hash_add(properties, DIFFUSION_TIME_SERIES_EVENT_VALUE_TYPE, "string");

TOPIC_SPECIFICATION_T *spec = topic_specification_init(TOPIC_TYPE_TIME_SERIES);
topic_specification_set_properties(spec, properties);

add_topic_from_specification(session, topic_path, spec, callback);

time_t end_time = time(NULL) + seconds;

while(time(NULL) < end_time) {

        const time_t time_now = time(NULL);
        const char *time_str = ctime(&time_now);

        BUF_T *value = buf_create();
        write_diffusion_string_value(time_str, value);

        /*
         *  Compose the timestamp value in milliseconds for the append
         */
        const long time_previous_day_ms = (time_now - (24 * 60 * 60)) * 1000;

        DIFFUSION_TIME_SERIES_TIMESTAMP_APPEND_PARAMS_T params = {
                .topic_path = topic_path,
                .datatype = DATATYPE_STRING,
                .value = value,
                .timestamp = time_previous_day_ms
        };

        diffusion_time_series_timestamp_append(session, params, NULL);
}

/*
 *  Subscribe to time series topics.
*/
static int on_value(const char *const topic_path,
             const TOPIC_SPECIFICATION_T *const specification,
             DIFFUSION_DATATYPE datatype,
             const DIFFUSION_VALUE_T *const old_value,
             const DIFFUSION_VALUE_T *const new_value,
             void *context)
{
        DIFFUSION_TIME_SERIES_EVENT_T *old_event;
        DIFFUSION_TIME_SERIES_EVENT_T *event;

        DIFFUSION_VALUE_T *old;
        DIFFUSION_VALUE_T *value;

        char *old_result;
        char *result;

        if(old_value != NULL) {
                read_diffusion_time_series_event(old_value, &old_event, NULL)

                old = diffusion_time_series_event_get_value(old_event);
                read_diffusion_string_value(old, &old_result, NULL);
        }

        read_diffusion_time_series_event(new_value, &event, NULL);

        char *author = diffusion_time_series_event_get_author(event);

        value = diffusion_time_series_event_get_value(event);
        read_diffusion_string_value(value, &result, NULL);

        return HANDLER_SUCCESS;
}

DIFFUSION_SESSION_FACTORY_T *session_factory = diffusion_session_factory_init();
diffusion_session_factory_principal(session_factory, "admin");
diffusion_session_factory_password(session_factory, "");

SESSION_T *session = session_create_with_session_factory(session_factory, "ws://::");

char *topic_path = "time-series-topic";

VALUE_STREAM_T *value_stream = calloc(1, sizeof(VALUE_STREAM_T));
value_stream->datatype = DATATYPE_STRING;
value_stream->on_value = on_value;

char *selector = malloc(1 + strlen(topic_path) + 1);
sprintf(selector, ">%s", topic_path);

add_time_series_stream(session, selector, value_stream);

SUBSCRIPTION_PARAMS_T params = {
        .topic_selector = selector,
        .on_topic_message = NULL
};

subscribe(session, params);

Get Started Today!

On-Premise, Cloud, SaaS or Hybrid deployment options to suit your IT Strategy

Easy to Download, Deploy and Operate

Download Free

Developers Home Everything you need to build applications using Diffusion

Diffusion SDK's

INDUSTRY SOLUTIONS

Technology Solutions

RESOURCES

Diffusion® Intelligent Event Data Platform

90%

1M Connections

1M Topics

Bring new capabilities to your applications

DATA GATEWAY

DATA WRANGLING

DATA DISTRIBUTION

What We Provide Our Customers

Bring new possibilities to your application

PUB-SUB

REQUEST-RESPONSE

TIME SERIES (EVENT HISTORY)

Store real-time events and dynamically query historical data enabling you to replay past events at any point in time.

FETCH

DELTAS

FLOW CONTROL

THROTTLING/CONFLATION

TOPIC MANAGEMENT

DATA AGNOSTIC

PERSISTED

DECLARATIVE LANGUAGE - DSL

LOW CODE CAPABILITIES

TRANSFORMATION

REST API

MQTT

PRE-BUILT ADAPTERS

RICH CLIENTS

AUTH INTEGRATION

DYNAMIC ACL

TTL

CLUSTERING

REMOTE SERVERS

EDGE REPLICATION

METRICS

DASHBOARD

CONTROL API

REMOTE SERVERS

REMOTE TOPIC VIEWS

CLUSTER-AWARE MESSAGING

PRE-BUILT ADAPTERS

BESPOKE INTEGRATIONS

REAL-TIME FIRE HOSES

How our Product Works

Our Customers

Yann L’Huillier

Our Customers

Ed Braidwood

Our Customers

Philipp Roeben

Simplicity in Action

Publish

Subscribe

Publish

Subscribe

Publish

Subscribe

Publish

Subscribe

Publish

Subscribe

Publish

Subscribe

Publish

Subscribe

Publish

Subscribe

Publish

Subscribe

Publish

Subscribe

Publish

Subscribe

Publish

Subscribe