For Developers

// 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');

//Subscribe to topics.
var session = Diffusion.Sessions
    .Principal("admin")
    .Password("")
    .Open("ws://::");
    
session.Topics.AddStream("my-topic", new ExampleValueStream());

await session.Topics.SubscribeAsync("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
     }
}

// Subscribe to topics.
final Session session = Diffusion.sessions()
  .principal("admin")
  .password("")
  .open("ws://:");

session.feature(Topics.class).addStream("my-topic", JSON.class, new Topics.ValueStream.Default() {
  @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-topic");

// 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) }

// Subscribe to topics.
final Session session = Diffusion.sessions()
  .principal("admin")
  .password("")
  .open("ws://:");

session.feature(Topics.class).addStream("my-topic", JSON.class, new Topics.ValueStream.Default() {
  @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-topic");

/*
 *  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);

# Subscribe to topics.
def on_update(*, old_value, topic_path, topic_value, **kwargs):
    print("Topic:", topic_path)
    if old_value is None:
        print("  Initial value:", topic_value)
    else:
        print("  Value updated")
        print("    Old value:", old_value)
        print("    New value:", topic_value)


def on_subscribe(*, topic_path, **kwargs):
    print(f"Subscribed to {topic_path}")


def on_unsubscribe(*, reason, topic_path, **kwargs):
    print(f"Unsubscribed from {topic_path} because {str(reason)}")


topic_selector = "foo/bar"
topic_type = diffusion.datatypes.STRING

session_duration = 15

value_stream = diffusion.topics.ValueStreamHandler(
    data_type=topic_type,
    update=on_update,
    subscribe=on_subscribe,
    unsubscribe=on_unsubscribe,
)

async def main():

    async with diffusion.Session(
        url="wss://:", principal="admin", credentials=diffusion.Credentials("")
    ) as session:

    print("Adding value stream")
    session.topics.add_value_stream(
        topic_selector=topic_selector, stream=value_stream
    )

    print(f"Subscribing to {topic_selector}")
    await session.topics.subscribe(topic_selector)
    await asyncio.sleep(session_duration)

    print(f"Unsubscribing from {topic_selector}")
    await session.topics.unsubscribe(topic_selector)