var handler = {
    onResponse : function(sessionID, response) {},
    onResponseError : function(sessionID, error) {},
    onError : function(error) {}
}
control.messages.sendRequestToFilter(filter, 'foo', 'Hello clients', handler, diffusion.datatypes.json(), diffusion.datatypes.json());

//Establish control sesssion
    final Session control = Diffusion.sessions().principal("control").password("password").open("ws://localhost:8080");
    
    //Obtain the MessagingControl feature
    final MessagingControl messagingControl = control.feature(MessagingControl.class);
    
    //Create a JSON object to send as a request
    final JSON request = Diffusion.dataTypes().json().fromJsonString("\"hello\"");
    
    //Send the request to a message path "foo", to all sessions which do not have a 'control' principal and wait for (at most) 5 seconds until the response (number of responses) is received.
    final int numberOfResponses = messagingControl.sendRequestToFilter("$Principal NE 'control'", "foo", request, JSON.class, JSON.class).get(5, TimeUnit.SECONDS);

session.messages.send('message_path', filter_query);

[session.messagingControl sendToFilter:filter
                                  path:message_path
                               message:[[PTDiffusionBytes alloc] initWithData:data]
                     completionHandler:^(NSUInteger count, NSError* error)
{
    if (error) {
        NSLog(@"Failed to send to %@. Error: %@", message_path, error);
    } else {
        NSLog(@"Sent to %lu sessions", (unsigned long)count);
    }
}];
                    

var options = session.MessagingControl.CreateSendOptionsBuilder().SetHeaders( headers ).Build();
session.MessagingControl.SendToFilter( filter_query, message_path, content, options, send_callback );

/*
 * Parameters for send_msg_to_session() call.
 */
SEND_MSG_TO_FILTER_PARAMS_T params = {
        .topic_path = message_path,
        .filter = filter,
        .content = message_content,
        .options.headers = headers,
        .options.priority = CLIENT_SEND_PRIORITY_NORMAL,
        .on_send = on_send_callback,
        .context = context
};

/*
 * Send the message and wait for the callback to acknowledge delivery.
 */
send_msg_to_filter(session, params);

// Create with a default listener function
session.messages.listen('message_path', function(message) {
    // Do something with the message
});

@interface MessageStreamDelegate : NSObject <PTDiffusionMessageStreamDelegate>
@end

@implementation MessageStreamDelegate

-(void)          diffusionStream:(PTDiffusionStream *)stream
    didReceiveMessageOnTopicPath:(NSString *)path
                         content:(PTDiffusionContent *)content
                         context:(PTDiffusionReceiveContext *)context
{
    // Do something when a message is received.
}

-(void)diffusionDidCloseStream:(PTDiffusionStream *)stream
{
    // Do something if the stream closes.
}

-(void)diffusionStream:(PTDiffusionStream *)stream didFailWithError:(NSError *)error
{
    // Do something if the stream fails.
}

// ... in later code 

// Ensure to maintain a strong reference to your message stream delegate as it
// is referenced weakly by the Diffusion client library.
MessageStreamDelegate *const delegate = [MessageStreamDelegate new];

// Create a locally evaluated topic selector specifying the messaging paths that
// should be captured by the stream.
PTDiffusionTopicSelector *const topicSelector =
    [PTDiffusionTopicSelector topicSelectorWithExpression:message_path];

// Use the Messaging feature to add a local stream using your delegate against
// the topic selector.
[session.messaging addMessageStreamWithSelector:topicSelector
                                       delegate:delegate];

session.feature(Messaging.class).addMessageStream(message_path, stream);

session.Messaging.AddMessageStream( message_path, stream );

/*
 * Register a listener for messages on the given path.
 */
MSG_LISTENER_REGISTRATION_PARAMS_T listener_params = {
        .topic_path = message_path,
        .listener = on_stream_message,
        .context = context
};
register_msg_listener(session, listener_params);

// Ensure to maintain a strong reference to your message stream delegate as it
// is referenced weakly by the Diffusion client library.
MessageStreamDelegate *const delegate = [MessageStreamDelegate new];

// Use the Messaging feature to add a local fallback stream using your delegate.
[session.messaging addFallbackMessageStreamWithDelegate:delegate];

messaging.addFallbackMessageStream(message_stream);

session.Messaging.AddFallbackMessageStream( stream );

 /*
 * Register a listener for any other messages.
 * (.topic_path is NULL).
 */
MSG_LISTENER_REGISTRATION_PARAMS_T global_listener_params = {
        .listener = on_stream_message,
        .context = context
};
register_msg_listener(session, global_listener_params);