Publish
session.topics.add('my-topic',diffusion.topics.TopicType.JSON);
var value = { foo : 'bar' };
var result = session.topicUpdate.set('my-topic', diffusion.datatypes.json(), value);await session.TopicControl.AddTopicAsync( “my-topic”, TopicType.JSON );
var jsonDataType = Diffusion.DataTypes.JSON;
var value = jsonDataType.FromJSONString("{"foo": "bar"}");
var result = session.TopicUpdate.SetAsync(
"my-topic",
value);session.feature(TopicControl.class).addTopic(
"my-topic",
TopicType.JSON,
new TopicControl.AddCallback.Default()); final JSONDataType jsonDataType = Diffusion.dataTypes()
.json();
final JSON value = jsonDataType.fromJsonString(
"{"foo" : "bar"}"); final CompletableFuture result = session.feature(TopicUpdate.class).set(
"my-topic", JSON.class, value);char *topic_path = “my-topic”;
BUF_T *buf = buf_create();
write_diffusion_json_value("{"foo" : "bar"}", buf);
DIFFUSION_TOPIC_UPDATE_SET_PARAMS_T topic_update_params = {
.topic_path = topic_path,
.datatype = DATATYPE_JSON,
.update = buf
};
diffusion_topic_update_set(session, topic_update_params);session.feature(TopicControl.class).addTopic("my-topic", TopicType.JSON);
final JSONDataType jsonDataType = Diffusion.dataTypes().json(); final JSON value = jsonDataType
.fromJsonString("{"foo" : "bar"}"); final CompletableFuture result = session.feature(
opicUpdate.class).set("my-topic", JSON.class, value)- JavaScript
- .NET
- Android
- Apple
- Java
Subscribe
session.addStream('my-topic', diffusion.datatypes.json()).on('value', function(topic, specification, newValue, oldValue) {
console.log(`New value for ${path}: ${newValue.get()`);
session.select('my-topic');
});class ExampleValueStream : IValueStream
{
public void OnSubscription(string topicPath, ITopicSpecification specification)
{
Console.WriteLine($"Subscribed to: {topicPath}");
}
public void OnUnsubscription(string topicPath, ITopicSpecification specification, TopicUnsubscribeReason reason)
{
Console.WriteLine($"Unsubscribed from: {topicPath}");
}
public void OnValue(string topicPath, ITopicSpecification specification, IJSON oldValue, IJSON newValue)
{
Console.WriteLine($"{topicPath}: {newValue.ToJSONString()}");
}
public void OnClose()
{
// Not used
}
public void OnError(ErrorReason errorReason)
{
// Not used
}
}
session.Topics.AddStream("my-topic", new ExampleValueStream());
await session.GetTopicsFeature().SubscribeAsync("my-topic");
session.feature(Topics.class).addStream("my-counter", JSON.class, new Topics.ValueStream.Default <
JSON > () {
@Override
public void onValue(String topicPath, TopicSpecification topicSpec, JSON oldValue,
JSON newValue) {
System.out.println("New value for" + topicPath + ": " + newValue.toJsonString());
}
});
session.feature(Topics.class).subscribe("my-counter");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)
{
char *result;
to_diffusion_json_string(new_value, &result, NULL);
printf(“New value for %s: %s \n”, topic_path, result);
free(result);
return HANDLER_SUCCESS;
}
char *topic_selector = “>my-topic”;
VALUE_STREAM_T value_stream = {
.datatype = DATATYPE_JSON,
.on_value = on_value
};
add_stream(session, topic_selector, &value_stream);
SUBSCRIPTION_PARAMS_T params = {
.topic_selector = topic_selector
};
subscribe(session, params);session.feature(Topics.class).addStream("my-counter", JSON.class, new Topics.ValueStream.Default <
JSON > () {
@Override
public void onValue(String topicPath, TopicSpecification topicSpec, JSON oldValue,
JSON newValue) {
System.out.println("New value for" + topicPath + ": " + newValue.toJsonString());
}
});
session.feature(Topics.class).subscribe("my-counter");- JavaScript
- .NET
- Android
- Apple
- Java
