Diffusion^®,
the Intelligent
Event-Data Platform

final Session session = Diffusion.sessions() 
    .principal("control") 
    .password("password") 
    .open("ws://localhost:8080"); 

const session = await diffusion.connect({ 
    host      : "localhost", 
    port      : 8080, 
    principal : "control", 
    password  : "password" 
}); 

var session = Diffusion.Sessions 
    .Principal("control") 
    .Password("password") 
    .Open("ws://localhost:8080"); 

DIFFUSION_ERROR_T error = { 0 }; 
SESSION_T *session = session_create("ws://localhost:8080", "control", "password", NULL, NULL, &error); 

async with diffusion.Session( 
        url = "ws://localhost:8080", 
        principal = "control", 
        credentials = diffusion.Credentials("password") 
) as session; 

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

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

session.feature(TopicControl.class).add("my-topic", new TopicSpecification(TopicType.STRING)); 
session.feature(TopicUpdate.class).set("my-topic", String.class, "Hello, world");  

session.topics.add('my-topic', diffusion.topics.TopicType.STRING); 
session.topicUpdate.set('my-topic', diffusion.datatypes.string(), 'Hello, world'); 

session.TopicControl.AddTopicAsync("my-topic", TopicType.STRING); 
session.TopicUpdate.SetAsync("my-topic", "Hello, World"); 

TOPIC_SPECIFICATION_T *spec = topic_specification_init(TOPIC_TYPE_STRING); 
add_topic_from_specification(session, "my-topic", spec, NULL); 

BUF_T *buf = buf_create(); 
write_diffusion_string_value("Hello, world", buf); 

TOPIC_UPDATE_SET_PARAMS_T topic_update_params = { 
  .topicPath = "my-topic", 
  .datatype = DATATYPE_STRING, 
  .update = buf 
}; 

buf_free(buf); 

await session.topics.add_topic("my-topic", diffusion.datatypes.STRING) 
await session.topics.set_topic("my_topic", "Hello, world", specification=diffusion.datatypes.STRING) 

self.session!.topicControl.add( 
    withTopicPath: "my-topic", 
    type: PTDiffusionTopicType.String, 
    value: nil, 
    completionHandler: self.errorHandler 
) 

self.session!.topicUpdateControl.updater.update( 
    withTopicPath: "my-topic", 
    value: "Hello, World", 
    completionHandler: self.errorHandler) 

curl --request POST \ 
     --header "authorization: Bearer {token}" \ 
     --header "content-type: application/json" \ 
     --data '{"path" : "my-topic", "type" : "string"}' \ 
     http://localhost:8080/topics/add 

curl --request POST \ 
     --header "authorization: Bearer {token}" \ 
     --header "content-type: application/json" \ 
     --data '{"path": "my-topic", "type": "string", "value": "Hello, world"}' \ 
     http://localhost:8080/topics/set 

session.feature(Topics.class).addStream("my-topic", String.class, new Topics.ValueStream.Default() { 
    @Override 
    public void onValue(String topicPath, TopicSpecification topicSpec, String oldValue, String newValue) { 
        System.out.println("New value for " + topicPath + ": " + newValue); 
    } 
}); 

session.feature(Topics.class).subscribe("my-topic"); 

session.addStream('my-topic', diffuison.datatypes.string()) 
    .on('value', (path, spec, newValue, oldValue) => { 
        console.log(`New value for ${path}: ${newValue}`); 
    }); 

session.select('my-topic'); 

class MyStream : IValueStream 
{ 
    public void OnValue(string topicPath, ITopicSpecification spec, string oldValue, string newValue) 
    { 
        Console.WriteLine($"New value for {topicPath}: {newValue}"); 
    } 
} 

var myStream = new MyStream(); 
session.Topics.AddStream("my-topic", myStream); 
session.Topics.SubscribeAsync("my-topic"); 

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 *str; 
  read_value_string_value(new_value, &str, NULL); 
  printf("New value for %s: %s\n", topic_path, str); 
  free(str); 

  return HANDLER_SUCCESS; 
} 

... 

VALUE_STREAM_T value_stream = { 
  .datatype = DATATYPE_STRING, 
  .on_value = on_value 
}; 
 
add_stream(session, "my_topic", &value_stream); 
 
SUBSCRIPTION_PARAMS_T params = { 
  .topic_selector = "my-topic" 
}; 

subscribe(session, params); 

def on_update(*, old_value, topic_path, new_value): 
    print("New value for " + topic_path + " = " + new_value) 

value_stream = diffusion.topics.ValueStreamHandler( 
    data_type = diffusion.datatypes.STRING 
    update = on_update 
) 

session.topics.add_value_stream( 
    topic_selector = "my-topic", stream = value_stream) 
} 

await session.topics.subscribe("my-topic") 

class StreamDelegate: NSObject, PTDiffusionJSONValueStreamDelegate { 
    func diffusionStream(_ stream: PTDiffusionValueStream, 
                         didUpdateTopicPath topicPath: String, 
                         specification: PTDiffusionTopicSpecification, 
                         oldString: String?, 
                         newString: String) { 
        print("New value for \(topicPath): \(newString)") 
    } 
} 

// ... 

let delegate = StreamDelegate() 

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

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

curl --request GET \ 
     --header "authorization: Bearer {token}" \ 
     --data '{"path": "my-topic", "type": "string", "value": "Hello, world"}' \ 
     http://localhost:8080/topics/fetch?path=my-topic