WebSockets with Red5 Pro

WebSockets allows you to connect your JavaScript code on client-side to server-side Java code and create a low-latency remote method invocation or push notification mechanism. For example: you could create your own Red5 Pro Java application with business logic for real-time communication, data gathering and more, and access the methods from client-side using JavaScript. Quite useful! Especially since this is a low-latency technique that can be used in conjunction with the very low-latency video streaming on Red5 Pro Server. Here are few tips for using WebSockets on Red5 Pro.

Traditional systems of using Flash for a chat application involved connecting the Flash client to the application via a particular scope in Red5 Pro. A scope is a logical separation within Red5 Pro, much like logical partitions on a physical hard drive. Scopes give you the advantage of better management of resources when building applications which involve lots of connections. Traditionally scopes were used as “rooms” in chat applications.

The standard way of connecting to a Red5 Pro application via a RTMP client is to use the following RTMP URL format: rtmp://host:5080/{application}. And to connect to a scope within the application, the URL format would be: rtmp://host:5080/{application}/{scopename}

WebSockets do not have rooms or scopes. So ws://localhost:8081/chat/room1/room2 won't work out of the box. Think of WebSockets as a single-level application where there is no depth. However there can be paths.

Initializing WebSockets For a Red5pro Application

Websockets can be enabled for a Red5pro application through its Application adapter class. The recommended way of doing this is to register websocket for the application in the appStart handler and remove it using the appStop handler. The code snippet given below shows how the Application adapter of the chat application is used for registering and unregistering with the websocket plugin.


public class Application extends MultiThreadedApplicationAdapter implements ApplicationContextAware {

    private static Logger log = Red5LoggerFactory.getLogger(Application.class, "chat");

    private ApplicationContext applicationContext;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    @Override
    public boolean appStart(IScope scope) {
        log.info("Chat starting");
        // get the websocket plugin
        WebSocketPlugin wsPlugin = (WebSocketPlugin) PluginRegistry.getPlugin("WebSocketPlugin");
        // add this application to it
        wsPlugin.setApplication(this);
        // get the manager
        WebSocketScopeManager manager = wsPlugin.getManager(scope);
        // get the ws scope
        WebSocketScope defaultWebSocketScope = (WebSocketScope) applicationContext.getBean("webSocketScopeDefault");
        // add the ws scope
        manager.addWebSocketScope(defaultWebSocketScope);
        return super.appStart(scope);
    }

    @Override
    public void appStop(IScope scope) {
        log.info("Chat stopping");
        // remove our app
        WebSocketScopeManager manager = ((WebSocketPlugin) PluginRegistry.getPlugin("WebSocketPlugin")).getManager(scope);
        manager.removeApplication(scope);
        manager.stop();
        super.appStop(scope);
    }
}

Once the application is registered with the websocket plugin, the next step is to create your websocket handler class, extending the org.red5.net.websocket.listener.WebSocketDataListener class. This class wil handle the standard server side websocket events for the clients. An example of the implementation would be the WebSocketChatDataListener.java class.

Finally we tell the application to use this WebSocketDataListener implementation to handle all websocket requests to our Red5pro application. This is done by adding a Bean definition to the context file (red5-web.xml) of the Red5pro application.

If you look at the Red5 red5-web.xml file of the red5 websocket chat sample application, you will see how a reference of the Red5 application is made available to the websocket listener via the virtual router (Router.java) using spring bean configuration.

    <bean id="web.handler" class="org.red5.demos.chat.Application" />

    <bean id="router" class="org.red5.demos.chat.Router">
        <property name="app" ref="web.handler" />
    </bean>

    <!-- WebSocket scope with our listeners -->
    <bean id="webSocketScopeDefault" class="org.red5.net.websocket.WebSocketScope" lazy-init="true">
        <!-- Application scope -->
        <constructor-arg ref="web.scope" />
        <!-- The ws scope listeners -->
        <property name="listeners">
            <list>
                <bean id="chatListener" class="org.red5.demos.chat.WebSocketChatDataListener">
                    <property name="router" ref="router" />
                </bean>
            </list>
        </property>
    </bean>

You can use this to form a means of communication between your Red5 application adapter and WebSocket data listener classes.

Using Scopes with WebSockets

To connect a WebSocket to a Red5 Pro application the following URL syntax is used: ws://{host}:8081/{application}. You then pass the scope path to the application via query string or URL path. How you wish to do this depends on your application design.


OPTION 1

rtmp://{host}:1935/{application}?scope=conference1
ws://{host}:8081/{application}?scope=conference1
  • When you use this method you will need to parse out the scope variable value from querystring.
  • Your clients will all be connecting to the top-most application level.
  • You need to design a mechanism to generate a unique name for your shared objects using the scope requested (since all connections are on single level). You then need to check if the shared object exists, and if not then create it with the evaluated name.
  • RTMP clients can push messages to shared objects via client-side API.
  • This will require special logic for WebSocket clients to resolve a SharedObject using scope name on server side and then push messages to it.

OPTION 2

rtmp://{host}:1935/{application}/conference1
ws://{host}:8081/{application}/conference1

When you use this method you will need to capture the path from the connection url for websocket handler. RTMP publisher will automatically create its sub scope(s).

WebSocketConnection.getPath()
  • Your RTMP clients will all be connecting to the scope specified in the RTMP URL and WebSocket connections will connect as they normally do.
  • RTMP clients can use same shared object name since the scope automatically manages isolation of shared object by same name. ie: /conference => SO and /conference1/SO are automatically separated and uniquely identified using the scope path.
  • RTMP clients can push messages to shared objects via client side API.
  • This will require special logic for WebSocket clients to resolve a SharedObject using scope name on server side and then push messages to it.

Or, you can also use a mix of Option 1 and 2 :

OPTION 3

rtmp://{host}:1935/{application}/conference1
ws://{host}:8081/{application}?scope=conference1

OPTION 4

rtmp://{host}:1935/{application}?scope=conference1
ws://{host}:8081/{application}/conference1

No matter which option you choose the challenge lies in connecting the WebSocket client to a scope for sending/receiving messages to/from RTMP clients. The answer to this can be found in the virtual router implementation Router.java of the sample WebSocket app red5 websocket chat.

Below is the function adapted from the red5-websocket-chat example app:

    /**
     * Get the Shared object for a given path.
     *
     * @param path
     * @return the shared object for the path or null if its not available
     */
    private ISharedObject getSharedObject(String path, String soname)
    {
        // get the application level scope
        IScope appScope = app.getScope();
        // resolve the path given to an existing scope
        IScope scope = ScopeUtils.resolveScope(appScope, path);
        if (scope == null)
        {
            // attempt to create the missing scope for the given path
            if (!appScope.createChildScope(path))
            {
                log.warn("Scope creation failed for {}", path);
                return null;
            }
            scope = ScopeUtils.resolveScope(appScope, path);
        }
        // get the shared object
        ISharedObject so = app.getSharedObject(scope, soname);
        if (so == null)
        {
            if (!app.createSharedObject(scope, soname, false))
            {
                log.warn("Chat SO creation failed");
                return null;
            }
            // get the newly created shared object
            so = app.getSharedObject(scope, "chat");
        }
        // ensure the so is acquired and our listener has been added
        if (!so.isAcquired())
        {
            // acquire the so to prevent it being removed unexpectedly
            so.acquire(); // TODO in a "real" world implementation, this would need to be paired with a call to release when the so is no longer needed
            // add a listener for detecting sync on the so
            so.addSharedObjectListener(new SharedObjectListener(this, scope, path));
        }
        return so;
    }

Explanation

The function accepts a path which is the location of the scope the WebSocket client is interested in for messages. This can be parsed from a query string or the WebSocket path property (as mentioned above). The second parameter is the shared object name on which it wishes to convey messages. This name needs to be same for RTMP and WebSocket clients.

The function tries first to find the scope at the given location path. If it fails to find one it will attempt to create one. If at least one RTMP client is connected to the scope it will persist automatically, otherwise it will be lost.

Once we have a scope we attempt to connect to a shared object in it by the name soname. As with the scope, we have to force-create a new shared object if we can't find an existing one. Finally we acquire it and register a ISharedObjectListener on it. This is to receive a notification when an event occurs on the SharedObject. A SharedObjectListener is used to monitor events occuring on the acquired SharedObject. The typical logic here is to update a attribute on the shared object such that it automatically triggers a sync event to all listeners (including flash clients).

NOTE: As a good programming habit make sure to release the acquired object when you know it wont be used anymore.

The ISharedObjectListener implementation for this SharedObject would look like this:

private final class SharedObjectListener implements ISharedObjectListener
{

    private final Router router;

    private final IScope scope;

    private final String path;

    SharedObjectListener(Router router, IScope scope, String path) {
        log.debug("path: {} scope: {}", path, scope);
        this.router = router;
        this.scope = scope;
        this.path = path;
    }

    @Override
    public void onSharedObjectClear(ISharedObjectBase so) {
        log.debug("onSharedObjectClear path: {}", path);
    }

    @Override
    public void onSharedObjectConnect(ISharedObjectBase so) {
        log.debug("onSharedObjectConnect path: {}", path);
    }

    @Override
    public void onSharedObjectDelete(ISharedObjectBase so, String key) {
        log.debug("onSharedObjectDelete path: {} key: {}", path, key);
    }

    @Override
    public void onSharedObjectDisconnect(ISharedObjectBase so) {
        log.debug("onSharedObjectDisconnect path: {}", path);
    }

    @Override
    public void onSharedObjectSend(ISharedObjectBase so, String method, List<?> attributes) {
        log.debug("onSharedObjectSend path: {} - method: {} {}", path, method, attributes);
    }

    @Override
    public void onSharedObjectUpdate(ISharedObjectBase so, IAttributeStore attributes) {
        log.debug("onSharedObjectUpdate path: {} - {}", path, attributes);
    }

    @Override
    public void onSharedObjectUpdate(ISharedObjectBase so, Map<String, Object> attributes) {
        log.debug("onSharedObjectUpdate path: {} - {}", path, attributes);
    }

    @Override
    public void onSharedObjectUpdate(ISharedObjectBase so, String key, Object value) {
        log.debug("onSharedObjectUpdate path: {} - {} = {}", path, key, value);
        // route to the websockets if we have an RTMP connection as the originator, otherwise websockets will get duplicate messages
        if (Red5.getConnectionLocal() != null) {
            router.route(scope, value.toString());
        }
    }
}

In the above class SharedObjectListener, take note of the method onSharedObjectUpdate. On the SharedObject update event we check to make sure that only messages from RTMP clients are relayed to WebSocket Clients. To prevent duplicates, messages from WebSocket clients are not relayed. (If you wanted to send messages from WebSocket to WebSocket you could design a unicast/multicast solution to check certain parameters, such as IP address, and relay messages only to specific WebSocket connections).