push-gists

// Get a reference to the JSON DataType instance
var jsonType = diffusion.datatypes.json();

// DataTypes can also be accessed by name
var binaryType = diffusion.datatypes.get('binary'); 

push-gists

// Create a stream for JSON values
session.stream("foo").asType(jsonType).on({
    value : function(topic, specification, newValue, oldValue) {
      // Receive the new value of a topic, along with the previous value
    },
    subscribe : function(topic, specification) {
      // Notification of subscription
    },
    unsubscribe : function(topic, specification, reason) {
      // Notification of unsubscription

push-gists

var diffusion = require('diffusion');

connectToDiffusion();

function connectToDiffusion(){
    console.log('Connecting to Diffusion...');
    
    diffusion.connect({
        host : 'localhost',
        port: 8080,

push-gists

function createTopics(session){
    console.log('Adding Topics...');
	
    session.topics.add('twitter', new diffusion.topics.TopicSpecification(diffusion.topics.TopicType.JSON)).then(
        function(result) {
            console.log('JSON Topic Added : ' + result.topic);

            // Start the updates to the topic
            startStream(session);
        },

push-gists

function startStream(session){
    console.log('Starting Twitter Stream');
    
    var stream = T.stream('statuses/filter', { track: 'push' })
    
    stream.on('tweet', function (tweet) {
        
        session.topics.update('twitter', tweet); // Update 'twitter' topic with the JSON for each tweet.
    })
}

push-gists

<script src="diffusion.js"></script>

<script>
    connectToDiffusion();
</script>

push-gists

function subscribeTopics(session){
    var datatype = diffusion.datatypes.json(); // Set DataType to JSON
    
    session.subscribe('twitter').asType(datatype).on({
        value : function(topic, specification, newValue, oldValue) {
            newValue = newValue.get();
            
            console.log(newValue); // output JSON to console.
        }
    });

push-gists

const webhookUri = "https://hooks.slack.com/services/WEBHOOK_ID";

push-gists

diffusion.connect({
	host: "REAPPT_HOST", 
	port: "REAPPT_PORT", 
	principal: "control", 
	credentials : "password", 
}).then(function (session) { 
	// Connection success! 
	console.log('Connected to Reappt');
	
	slack.setWebhook(webhookUri); 

push-gists

function subscribeToTopic(session){ 
	// Set datatype to JSON 
	var datatype = diffusion.datatypes.json(); 
	
	session.subscribe('slack').asType(datatype).on({ 
		
		value : function(topic, specification, newValue, oldValue) { 
			newValue = newValue.get(); 
			
			console.log(newValue);