The Streams API

Qbix provides three layers of increasing abstraction above your database server:

1. The database abstraction layer
2. The database models
3. The Streams API

When building your apps, try to start with the highest abstraction layer — Streams — for implementing your data models. Before invoking the lower abstraction layers, make sure you really need to interface with the database more directly.

The Streams plugin exposes three robust interfaces: one for PHP scripts, one as a REST interface for HTTP clients, and one as part of the client-side Javascript SDK.

All an app has to do to get access to this API is include the Streams plugin in config/app.json:

{
  "Q": {
    "app": "First",
    ...
    "plugins": ["Users", "Streams"]
  },
  ...
}

The API takes care of a lot of things, including:

• syncing between the server and multiple clients
• posting, broadcasting messages in real time
• events for updating views on real time updates
• subscribing to offline notifications
• inviting other users to the stream
• calculating access from roles and permissions
• search and relations between streams
• scaling to billions of streams

Basic stream fields

Every stream has the following fields:

• publisherId: id of the user publishing the stream
• name: the name of the stream. By convention, it should begin with the type of the stream
• type: the type of the stream. For example "Websites/seo" or "Streams/user/firstName"
• title: the title of the stream, such as "Untitled Document" or "First Name"
• icon: relative or absolute URL of a directory containing the stream's icon images
• content: stores up to 1000 characters of human readable content or a short summary
• attributes: stores up to 1000 characters of JSON attributes
• readLevel, writeLevel, adminLevel: specify public access
• inheritAccess: if set, inherits access from another stream
• messageCount: index of last message posted to the stream
• participantCount: number of participants in the stream
• closedTime: if set, the stream is marked closed

The PHP API

Here is how you fetch streams from the database:

$streams = Streams::fetch(
  $asUserId, $publisherId, $name, $options
);

The first parameter, $asUserId, is used to determine the access levels on the streams obtained by this method. You can pass a user id here or a Users_User object. Passing the empty string "" means that the streams should be fetched in the form they are accessible to the public. Finally, passing null is a shortcut for Users::loggedInUser(), which causes the streams to be fetched as the currently logged-in user, or else as they appear to the public.

The second parameter, $publisherId, is the id of the user who is publishing the streams to be fetched. The streams should be hosted on a database that is reachable by a local database connection, which means that they are hosted on the internal computer network and not somewhere on some remote domain.

The third parameter, $name, can be an array of stream names, or a database range. You could also pass a string here that ends in a slash "/". This will fetch all the streams whose names begin with that string as a prefix. By convention, a stream's type is found in the prefix of its name, so this can be used to find all streams of a particular type. However, it is usually more appropriate to use relations and search instead.

The last parameter can take options such as "limit", "offset", "orderBy", "where", and "refetch".

The returned result is an array of $name => $stream pairs in the same order as the names provided in $name. (Streams not found in the database are represented by null). Streams that were previously fetched are not fetched again, but instead the already constructed Streams_Stream objects are returned. (That is, unless the "refetch" option is set to true.)

To fetch one stream from the database, you would do:

$stream = Streams::fetchOne(
  $asUserId, $publisherId, $name,
  $fields, $options
);

Please look at the reference for this function, for more options.

You can, of course, manually retrieve a stream like any other object extending Db_Row. As we stressed above, however, you should have a good reason for bypassing the Streams API:

$stream = new Streams_Stream();
$stream->publisherId = $userId;
$stream->name = "Websites/seo/38u9c9x";
$stream->retrieve();
$stream->calculateAccess($asUserId);

Besides the icon, title and content fields, streams have a JSON string of up to 1000 characters that can store a few additional attributes. Here is how you can manipulate a stream's attributes in PHP:

$attributes = $stream->getAttributes();
$foo = $stream->getAttribute('foo', $default);
$stream->setAttribute('foo', $foo);
$stream->clearAttribute('foo')

Some stream types may want to extend this basic structure for their needs.

Creating Streams, using Templates

You can create streams by calling the following:

$stream = Streams::create(
  $asUserId, $publisherId, $type,
  $fields, $relate
);

This either creates the stream, or throws Users_Exception_notAuthorized if the access checks determine the user doesn't have the authority to create this stream.

When a stream is created, the Streams plugin looks for a possible template, which is a stream whose name is your stream's type followed by a slash, (e.g. "Chess/game"). First it looks for a publisherId matching the one of the stream you want to create, and if not found, it also tries to look for a template where publisherId = "". If a template is found, then the fields of its row are automatically copied into the newly created stream, including the access control for the public. Also, any corresponding rows from streams_access, streams_subscription and streams_rule may be copied as well. (This means rows with same publisherId and streamName as the template that was found.)

This basically means that the system automatically copies access, subscription and notification rules for various types of streams as they are created. If the templates are subsequently changed, all existing streams remain as they are. If you ever want to change access to all existing streams of a certain type, all at once and going forward, check out the mutable access feature.

The HTTP API

The Streams plugin implements several actions to expose a REST interface for HTTP clients. Since this article is about streams, we will discuss the endpoint "Streams/stream", whose url is $baseUrl/action.php/Streams/stream

GET Streams/stream?publisherId=...&name=...
Handles dynamic requests that expect the slot "stream" to be filled. If the stream doesn't exist, this slot is filled with null, otherwise it is filled with the result of $stream->exportArray(). Clients can also request the "participants" and "messages" slots to be filled, and indicate the maximum number of each to return in the correspondingly named "participants" and "messages" querystring fields.

POST Streams/stream
Do this to create a new stream. Expects at least the following POST fields:

• publisherId: id of user publishing the stream
• type: the type of the stream to create

This may throw a Users_Exception_notAuthorized error if the logged-in user lacks authorization to create a stream. In addition, the error will be thrown unless the config allows the client to create this particular type of stream, by having "Streams"/"types"/$streamType/"create" config field be either true or an array naming the fields that can be set by the client. Similar rules apply when the client tries to edit existing streams via HTTP PUT requests, except with "edit" instead of "create". For existing stream types, the configs are already correctly set, but keep this in mind when designing your own stream types.

A unique stream name is automatically generated, of the form "$type/$randomString". Your app or plugin needs to implement its own actions if it wants to expose greater control over stream names to clients asking to create certain types of streams. Check out the Websites plugin, for instance.

PUT Streams/stream?publisherId=...&name=...
Use this to edit an existing stream. An exception is thrown if the stream was not found.

For both POST and PUT, you set various stream fields in the body of the request. In addition, you can pass fields named "attribute[attrName]" to set attributes.

Besides requiring authorization to edit a stream, you must have true in the config field "Streams"/"types"/$type/"edit" for each type of stream clients will be allowed to edit.

DELETE Streams/stream?publisherId=...&name=...
closes an existing stream. An exception is thrown if the stream was not found or the user is not authorized to delete the stream.

POST Streams/batch
This allows clients to send batched GET requests, to be executed in order. The requests can be for streams, messages, participants and avatars.

The Client Javascript API

The Streams plugin provides a Javascript API that you can use in your web apps. Under the hood, it makes great use of code patterns provided by Qbix, making your apps are more effective without you having to think about it.

Here is how you would get a stream:

 Q.Streams.get(pubId, streamName, callback);

This function (and others) are wrapped with Q.getter so you can use them every time you need a stream object, without worrying about unnecessary requests to the server, having to cache objects, or anything else. It even batches requests together.

You can also get a stream together with the latest n messages and p participants in it, like this:

Q.Streams.get(pubId, streamName, callback,
  { participants: 10, messages: 10 }
);

Here is how you would create streams

// Create a stream with given fields
// only publisherId and type is required
Q.Streams.create(fields, callback);

// Create a stream related to existing stream
Q.Streams.create(fields, callback, {
  publisherId: "a8z7c81",
  streamName: "Media/album/827c918c",
  type: "Media/photo"
});

// Construct a stream locally, to save later:
Q.Streams.construct(fields, callback);

In order for the server to allow clients to create streams, remember to set the "Streams"/"types"/$type/"create" config field to true.

The rest of the API is supposed to be just as intuitive. When you have a stream object you can do things with it, such as:

Q.Streams.get(pubId, name, function () {
  var stream = this;
  // set pending fields
  stream.set('content', "new content");
  stream.set('title', "new title");
  // set pending attribute
  stream.setAttribute(name, value);
  oldValue = stream.getAttribute(name);
  pendingVal = stream.getAttribute(name, true);
  stream.clearAttribute(name);
  // requests to the server
  stream.save(callback); // save pending
  stream.remove(callback);
});

Refresh, retain, release

Sometimes you think the stream may have changed on the server, and you want to refresh it:

// if you have the actual stream object:
stream.refresh(options);

// if you just have the publisherId and name:
Q.Streams.Stream.refresh(
  publisherId, streamName, callback, options
);

Other times, your app may "wake up" after being offline, or being in the background on your mobile phone, and you will want to make sure the user is seeing up-to-date information. One way, of course, is to just reload the page, but this usually causes flicker and things jumping around unnecessarily. In fact, cellphones often fire "online" and "offline" events while you are browsing the web, which will cause your page to reload constantly.

Instead, what you really want to do is refresh all the streams that are currently being used by the various tools and views. This will trigger all the right events and the views will update themselves only if necessary. Here's how you do it:

Q.Streams.refresh(callback, options); // shocking?

By default, the Streams plugin already starts out listening to events such as Q.onOnline and calls Q.Streams.refresh() automatically.

But how does the Streams plugin know what streams to refresh? When code in various tools and pages requests streams from the server, it can also retain and release them. Here is an example:

Q.page("First/welcome", function () {
  Q.Streams.retainWith(true)
  .get(publisherId, "Chess/game/ac817293",
    function () {
      var stream = this; 
      // do something with the stream
    }
  );
	
  // You do NOT need to do the below
  // because all streams retained by
  // the page are automatically released
  // when the page is about to be unloaded.
  // It is just for illustration purposes:
  return _beforeUnload() {
    Q.Streams.release(true);
  }
});

You don't necessarily want to retain every stream that you get from the server, so it's up to you to indicate when you would like to retain a stream. Here is how you would do it when designing a tool:

Q.Tool.define("Chess/game", function () {
  var tool = this;
  var publisherId = Q.Users.loggedInUser.id;
  Q.Streams.retainWith(tool)
  .get(publisherId, "Chess/game/ac817293",
    function () {
      tool.state.game = this;
	  tool.stateChanged('game');
      // do something
    }
  );
},
{}, // default tool options
{
  doSomething: function () {
    var game = this.state.game;
    // use the stream maybe
    game.refresh();
    game.move(from, to);
  },
	
  // You do NOT need to do the below
  // because all streams retained by
  // the tool are automatically
  // released when the tool is removed.
  // It is just for illustration purposes:
  Q: {
    beforeRemove: {"Chess/game": function () {
      Q.Streams.release(this);
    }}
  }
})

When a stream is released from all the pages, tools, and custom keys under which it was retained, then it is no longer on the list to be refreshed during Q.Streams.refresh().

Stream Constructors

Each stream type can have its own constructor just like tools do.

Client and Server

As you may have already guessed, Streams in Qbix are a way of modeling data with the server being the authority on the contents of the stream. The Streams API on the client uses Q.getter and Q.Cache to hold a cache of the data, but it relies on the server to tell it when the data has updated. This will become even more clear when we talk about posting messages to streams, since multiple clients can be writing to the stream and all of them need to see the messages posted in the same order.

Sometimes, when rendering a page, you would like to "preload" some streams, to avoid having to do another round-trip to the server later. The Streams API finds them in the response and caches them, so when Q.Streams.get() is called, it will just use the cached version right away, leading to a faster user experience. Here is how you would preload streams when generating the page:

// add to the list of preloaded streams
$stream->addPreloaded($asUserId);

// if you want to cancel the preloading
$stream->removePreloaded();

// (yeah, it's that straightforward)

Note, however, that this information will be sent every time the page is requested. You can avoid that by using $_SESSION (for example). If the preloaded stream is expelled from the cache later, then the next Q.Streams.get() attempting to get the stream will just ask the server for it again.

If you want to override the default error handler for catching server errors, check out the Q.Streams.onError event.

Closing a stream

Suppose you have a chatroom where people are discussing an upcoming event. When an event has already happened, you may want to "close" the chatroom, preserving its history. Or you may want to "close" comments for a blog post 30 days after it was published. "Closing" is the typical action that the Streams HTTP API exposes to clients during the DELETE verb.

Closing the stream does not delete it. There are several reasons for this, but the main one is that clients may be listening for messages on this stream, so it has to stick around until all the clients -- even the ones that have gone offline temporarily -- have been made aware it's been "closed". Otherwise, they'll just request the stream and get a "stream is missing" error. Other benefits to not deleting the stream right away include the ability to undo the deletion. If you really want to let clients delete streams, write your own HTTP action handler.

However, closed streams are not accessible anymore for any users except those who have at least the write level to close it in the first place. In addition, closing a stream removes all the "relations" so the stream no longer shows up in lists of "related streams". This cannot be easily undone since no backup of these relations is saved.