Diffusion^® Intelligent Data Mesh

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");

class StreamDelegate: NSObject, PTDiffusionJSONValueStreamDelegate {
        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")
           }
    }
}

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

var response = requests.post('https://api.diffusion.cloud/topics/fetch', headers={
'Authorization': 'Bearer ' + token }, json = { selector: '?sensors/.*' })
print (response.json()[‘results])

$client = new \GuzzleHttp\Client(["base_uri" => "https://api.diffusion.cloud"]);
$response = $client->post("/topics/fetch", [
 'headers' => ['Authorization' => 'Bearer '],
 'json' => [ 'selector' => '?sensors/.*']
]);
echo json_decode($response->getBody())[‘results’];

req, _ := http.NewRequest("POST", "https://api.diffusion.cloud/topics/fetch",
strings.NewReader("{\selector\": \"?sensors/.*\"}"))
req.Header.Add("Authorization", "Bearer ")
req.Header.Add("content-type", "application/json")
client := &http.Client {}
res, err := client.Do(req)

curl --include
  --request POST
  --url https://api.diffusion.cloud/topics/fetch
  --header 'authorization: Bearer '
  --header 'content-type: application/json'
  -d '{"selector": "?sensors/.*" }'