Skip to main content

Getting Started with Streams

This article is an introduction to using streams with Macrometa SDKs.

Prerequisites​

Get Started with Streams​

This page guides you through creating a stream, publishing messages to it, and subscribing to the stream using the pyC8 and jsC8 SDKs.

  1. Create a new JavaScript (.js) or Python (.py) file in your favorite IDE.
  2. Copy the code block below and paste it into your JavaScript or Python file.
  3. With each subsequent step, append the code block to the existing file and then run it.

If you want to skip the explanation and just run the code, then go directly to the Full Demo File.

Step 1. Connect to GDN​

To use streams with Macrometa Global Data Network (GDN), you must first establish a connection to a local region. When this code runs, it initializes the server connection to the specified region URL. For more information about connecting to GDN, refer to Authentication.

const jsc8 = require("jsc8");

// Choose one of the following methods to access the GDN. API key is recommended.
// API key
const client = new jsc8({url: "https://play.paas.macrometa.io", apiKey: "XXXX", fabricName: '_system'});

// JSON Web Token
// const client = new jsc8({url: "https://play.paas.macrometa.io", token: "XXXX", fabricName: '_system'});

// Or use email and password to authenticate client instance
// const client = new jsc8("https://play.paas.macrometa.io");
// Replace values with your email and password.
// await client.login("nemo@nautilus.com", "xxxxxx");

Step 2. Get GeoFabric Details​

Get fabric details, including the name and associated regions.

async function getFabric() {
  try {
    await console.log("Getting the fabric details...");
    let result = await client.get();

    await console.log("result is: ", result);
  } catch(e){
    await console.log("Fabric details could not be fetched because "+ e);
  }
}

getFabric();

Step 3. Create Global and Local Streams​

The streams in GDN can be either local or globally geo-replicated. The code below allows you to create either or both and then get the stream details.

async function streams() {
    try{
      await console.log("Creating local stream...");
      const stream_local = await client.createStream("testStream-local", true);

      await console.log("Creating global stream...");
      const stream_global = await client.createStream("testStream-global", false);

    } catch(e){
      await console.log("Streams could not be fetched because "+ e);
    }
}

streams();

Step 4. Publish Messages​

Example to publish documents to a stream. The stream can be either a local stream or could be a geo-replicated stream.

async function streams() {
    try {
      await console.log("Creating local stream...");
      const stream = client.stream("my-stream", true);
      await stream.createStream();
      const producerOTP = await stream.getOtp();
      const producer = await stream.producer("play.paas.macrometa.io", {
        otp: producerOTP,
    });
      producer.on("open", () => {
        // If your message is an object, convert the object to a string.
        // e.g. const message = JSON.stringify({message:'Hello World'});
        const message = "Hello World";
        const payloadObj = { payload: Buffer.from(message).toString("base64") };
        producer.send(JSON.stringify(payloadObj));
      });
      producer.on("message", (msg) => {
        console.log(msg, "Sent Successfully");
      });

    } catch(e) {
      await console.log("Publishing could not be done because "+ e);
    }
}

streams()

Step 5. Subscribe to Stream​

Example to subscribe documents from a stream. The stream can be either a local stream or a geo-replicated global stream.

async function getDCList() {
  const geo_fabric = "_system"
  let dcListAll = await client.listUserFabrics();
  let dcListObject = await dcListAll.find(function(o) { return o.name === geo_fabric; });
  return dcListObject.options.dcList.split(",");
}

(async function() {
  const dcList = await getDCList();
  await console.log("dcList: ", dcList);
  await client.createStream("my-stream", true);
  
  //Here the last boolean value tells if the stream is local or global. false means that it is global.
  const consumer = await client.createStreamReader("my-stream", "my-subscription", true);
  consumer.on("message", (msg) => {
    const { payload, messageId } = JSON.parse(msg);
    
    // Received message payload
    console.log(Buffer.from(payload, "base64").toString("ascii"));
    // Send message acknowledgement
    consumer.send(JSON.stringify({ messageId }));
  });

})();

Full Demo File​

It's time to see streams in action!

  1. Replace the contents of your .js or .py file from above with the code block below.
  2. In your browser, open the GDN console and then click Streams.
  • Select your recently created stream (c8globals.streamQuickstart) to view the output of the message within the console.
  1. Open two terminal windows and start the program in each window
  • In one terminal, type 'r' to begin listening for messages, while in the other terminal, type 'w' to begin writing messages
  • Upon each write, you should see the message received in the second terminal window, as well as the message displayed in the GDN console output
// Connect to GDN.
const jsc8 = require("jsc8");
// Choose one of the following methods to access the GDN. API key is recommended.
const client = new jsc8({url: "https://play.paas.macrometa.io", apiKey: "xxxxx", fabricName: '_system'});
console.log("Authentication done!!...");

// Get GeoFabric details.
async function getFabric() {
    try {
      await console.log("Getting the fabric details...");
      let result = await client.get();
  
      await console.log("result is: ", result);
    } catch(e) {
      await console.log("Fabric details could not be fetched because "+ e);
    }
}

// Create global and local streams.
async function createStreams() {
    try{
      console.log("Creating local stream...");
      await client.createStream("testStream-local", true);

      console.log("Creating global stream...");
      await client.createStream("testStream-global", false);

    } catch(e) {
      await console.log("Streams could not be created because "+ e);
    }
}

// Subscribe to stream
async function createConsumer() {
  const dcList = await getDCList();
  await console.log("dcList: ", dcList);

  try{
    console.log("Creating local stream...");
    await client.createStream("my-stream", true);
  } catch(e) {
    await console.log("Stream could not be created because "+ e);
  }

  // Here the last boolean value tells if the stream is local or global. false means that it is global.
  const consumer = await client.createStreamReader("my-stream", "my-subscription", true);

  consumer.on("message", (msg) => {
    const { payload, messageId } = JSON.parse(msg);
    
    // Received message payload
    console.log(Buffer.from(payload, "base64").toString("ascii"));

    // Send message acknowledgement
    consumer.send(JSON.stringify({ messageId }));
  });
}

// Publish messages to stream.
async function createProducer() {
    try {
      await console.log("Creating local stream...");
      const stream = client.stream("my-stream", true);
      const producerOTP = await stream.getOtp();
      const producer = await stream.producer("play.paas.macrometa.io", {
        otp: producerOTP,
      });
      producer.on("open", () => {
        // If your message is an object, then convert the object to a string.
        // e.g. const message = JSON.stringify({message:'Hello World'});
        const message = "Hello World";
        const payloadObj = { payload: Buffer.from(message).toString("base64") };
        producer.send(JSON.stringify(payloadObj));
      });
      producer.on("message", (msg) => {
        console.log(msg, "Sent successfully");
      });
    } catch(e) {
      await console.log("Publishing could not be done because "+ e);
    }
}

async function getDCList() {
    const geo_fabric = "_system"
    let dcListAll = await client.listUserFabrics();
    let dcListObject = await dcListAll.find(function(o) { return o.name === geo_fabric; });
    return dcListObject.options.dcList.split(",");
}

async function main() {
    await getFabric();
    await createStreams();
    await createConsumer();
    await createProducer();
};

main();