Chapter 5
Event Notification Service API Reference
This chapter details the ENS API; it is divided into four main sections:
Publisher API Functions List
This chapter includes a description of the following Publisher functions, listed in Table 5-1:
Subscriber API Functions List
This chapter includes a description of following Subscriber functions, listed in Table 5-2:
Publish and Subscribe Dispatcher Functions List
This chapter includes a description of the following Publish and Subscribe Dispatcher functions, listed in Table 5-3:
Table 5-3 ENS Publish and Subscribe Dispatcher Functions List
|
Definition/Function
|
Description
|
|
pas_dispatcher_t
|
Definition of a publish and subscribe dispatcher.
|
|
pas_dispatcher_new
|
Creates a dispatcher.
|
|
pas_dispatcher_delete
|
Destroys a dispatcher created with pas_dispatcher_new.
|
|
pas_dispatch
|
Starts the dispatch loop of an event notification environment.
|
|
pas_shutdown
|
Stops the dispatch loop on an event notification environment started with pas_dispatch.
|
Publisher API
The Publisher API consists of one definition and nine functions:
publisher_t
Purpose.
A publisher.
Syntax
typedef struct enc_struct publisher_t;
Parameters
None.
Returns
Nothing.
publisher_cb_t
Purpose.
Generic callback function invoked by ENS to acknowledge an asynchronous call.
Syntax
typedef void (*publisher_cb_t) (void *arg, int rc, void *data);
|
arg
|
Context variable passed by the caller.
|
|
rc
|
The return code.
|
|
data
|
For an open, contains a newly created context.
|
">
Parameters
|
arg
|
Context variable passed by the caller.
|
|
rc
|
The return code.
|
|
data
|
For an open, contains a newly created context.
|
Returns
Nothing.
publisher_new_a
Purpose
Creates a new asynchronous publisher.
Syntax
void publisher_new_a (pas_dispatcher_t *disp,
void *worker,
const char *host,
unsigned short port,
publisher_cb_t cbdone,
void *cbarg);
|
disp
|
P&S thread pool context returned by pas_dispatcher_new.
|
|
worker
|
Application worker. If not NULL, grouped with existing workers created by ENS to service this publisher session. Used to prevent multiple threads from accessing the publisher data at the same time.
|
|
host
|
Notification server host name.
|
|
port
|
Notification server port.
|
|
cbdone
|
The callback invoked when the publisher has been successfully create, or could not be created.
There are three Parameters to cbdone:
cbarg
The first argument.
A status code.
If non-zero, the publisher could not be created; value specifies cause of the failure.
The new active publisher.
|
|
cbarg
|
First argument of cbdone.
|
">
Parameters
|
disp
|
P&S thread pool context returned by pas_dispatcher_new.
|
|
worker
|
Application worker. If not NULL, grouped with existing workers created by ENS to service this publisher session. Used to prevent multiple threads from accessing the publisher data at the same time.
|
|
host
|
Notification server host name.
|
|
port
|
Notification server port.
|
|
cbdone
|
The callback invoked when the publisher has been successfully create, or could not be created.
There are three Parameters to cbdone:
cbarg
The first argument.
A status code.
If non-zero, the publisher could not be created; value specifies cause of the failure.
The new active publisher.
|
|
cbarg
|
First argument of cbdone.
|
Returns
Nothing. It passes the new active publisher as third argument of cbdone callback.
publisher_new_s
Purpose
Creates a new synchronous publisher.
Syntax
publisher_t *publisher_new_s (pas_dispatcher_t *disp,
void *worker,
const char *host,
unsigned short port);
|
disp
|
P&S thread pool context returned by pas_dispatcher_new
|
|
worker
|
Application worker. If not NULL, grouped with existing workers created by ENS to service this publisher session. Used to prevent multiple threads from accessing the publisher data at the same time.
|
|
host
|
Notification server host name.
|
|
port
|
Notification server port.
|
">
Parameters
|
disp
|
P&S thread pool context returned by pas_dispatcher_new
|
|
worker
|
Application worker. If not NULL, grouped with existing workers created by ENS to service this publisher session. Used to prevent multiple threads from accessing the publisher data at the same time.
|
|
host
|
Notification server host name.
|
|
port
|
Notification server port.
|
Returns
A new active publisher (publisher_t).
publish_a
Purpose
Sends an asynchronous notification to the notification service.
Syntax
void publish_a (publisher_t *publisher,
const char *event_ref,
const char *data,
unsigned int datalen,
publisher_cb_t cbdone,
publisher_cb_t end2end_ack,
void *cbarg,
unsigned long timeout);
|
publisher_t
|
The active publisher.
|
|
event_ref
|
The event reference. This is a URI identifying the modified resource.
|
|
data
|
The event data. The body of the notification message. It must be MIME object. It is opaque to the notification service, which merely relays it to the events' subscriber.
|
|
datalen
|
The length in bytes of the data.
|
|
cbdone
|
The callback invoked when the data has been accepted or deemed unacceptable by the notification service. What makes a notification acceptable depends on the protocol used. The protocol may choose to use the transport acknowledgment (TCP) or use its own acknowledgment response mechanism.
|
|
end2end_ack
|
The callback function invoked after acknowledgment from the consumer peer (in an RENL) has been received. Used only in the context of an RENL.
|
|
cbarg
|
The first argument of cbdone or end2end_ack when invoked.
|
|
timeout
|
The length of time to wait for an RENL to complete.
|
">
Parameters
|
publisher_t
|
The active publisher.
|
|
event_ref
|
The event reference. This is a URI identifying the modified resource.
|
|
data
|
The event data. The body of the notification message. It must be MIME object. It is opaque to the notification service, which merely relays it to the events' subscriber.
|
|
datalen
|
The length in bytes of the data.
|
|
cbdone
|
The callback invoked when the data has been accepted or deemed unacceptable by the notification service. What makes a notification acceptable depends on the protocol used. The protocol may choose to use the transport acknowledgment (TCP) or use its own acknowledgment response mechanism.
|
|
end2end_ack
|
The callback function invoked after acknowledgment from the consumer peer (in an RENL) has been received. Used only in the context of an RENL.
|
|
cbarg
|
The first argument of cbdone or end2end_ack when invoked.
|
|
timeout
|
The length of time to wait for an RENL to complete.
|
Returns
Nothing.
publish_s
Purpose
Sends a synchronous notification to the notification service.
Syntax
int publish_s (publisher_t *publisher,
const char *event_ref,
const char *data,
unsigned int datalen);
|
publisher
|
The active publisher.
|
|
event_ref
|
The event reference. This is a URI identifying the modified resource.
|
|
data
|
The event data. The body of the notification message. It must be a MIME object. It is opaque to the notification service, which relays it to the events' subscriber.
|
|
datalen
|
The length in bytes of the data.
|
">
Parameters
|
publisher
|
The active publisher.
|
|
event_ref
|
The event reference. This is a URI identifying the modified resource.
|
|
data
|
The event data. The body of the notification message. It must be a MIME object. It is opaque to the notification service, which relays it to the events' subscriber.
|
|
datalen
|
The length in bytes of the data.
|
Returns
Zero if successful; a failure code if unsuccessful. If an RENL, the call does not return until the consumer has completely processed the notification and has successfully acknowledged it.
publisher_delete
Purpose
Terminates a publish session.
Syntax
void publisher_delete (publisher_t *publisher);
|
publisher
|
The publisher to delete.
|
">
Parameters
|
publisher
|
The publisher to delete.
|
Returns
Nothing.
publisher_get_subscriber
Purpose
Creates a subscriber using the credentials of the publisher.
Syntax
struct subscriber_struct * publisher_get_subscriber(publisher_t
*publisher);
|
publisher
|
The publisher whose credentials are used to create the subscriber.
|
">
Parameters
|
publisher
|
The publisher whose credentials are used to create the subscriber.
|
Returns
The subscriber, or NULL if the creation failed. If the creation failed, use the subscriber_new to create the subscriber.
renl_create_publisher
Purpose
Declares an RENL, which enables the end2end_ack invocation. After this call returns, the end2end_ack argument is invoked when an acknowledgment notification matching the specified publisher and subscriber is received.
Syntax
void renl_create_publisher (publisher_t *publisher,
const char *renl_id,
const char *subscriber,
publisher_cb_t cbdone,
void *cbarg);
|
publisher
|
The active publisher.
|
|
renl_id
|
The unique RENL identifier. This allows two peers to be able to set up multiple RENL's between them.
|
|
subscriber
|
The authenticated identity of the peer.
|
|
cbdone
|
The callback invoked when the RENL is established.
|
|
cbarg
|
The first argument of cbdone, when invoked.
|
">
Parameters
|
publisher
|
The active publisher.
|
|
renl_id
|
The unique RENL identifier. This allows two peers to be able to set up multiple RENL's between them.
|
|
subscriber
|
The authenticated identity of the peer.
|
|
cbdone
|
The callback invoked when the RENL is established.
|
|
cbarg
|
The first argument of cbdone, when invoked.
|
Returns
Nothing.
renl_cancel_publisher
Purpose
This cancels an RENL. This does not prevent more notifications being sent, but should a client acknowledgment be received, the end2end_ack argument of publish will no longer be invoked. All RENL's are automatically destroyed when the publisher is deleted. Therefore, this function does not need to be called to free RENL-related memory before deleting a publisher.
Syntax
void renl_cancel_publisher (renl_t *renl);
|
renl
|
The RENL to cancel.
|
">
Parameters
Returns
Nothing.
Subscriber API
The Subscriber API includes two definitions and ten functions:
subscriber_t
Purpose
A subscriber.
Syntax
typedef struct enc_struct subscriber_t;
Parameters
None.
Returns
Nothing.
subscription_t
Purpose
A subscription.
Syntax
typedef struct subscription_struct subscription_t;
Parameters
None.
Returns
Nothing.
subscriber_cb_t
Purpose
Generic callback function invoked by ENS to acknowledge an asynchronous call.
Syntax
typedef void (*subscriber_cb_t) (void *arg,
int rc,
void *data);
|
arg
|
Context variable passed by the caller.
|
|
rc
|
The return code.
|
|
data
|
For an open, contains a newly created context.
|
">
Parameters
|
arg
|
Context variable passed by the caller.
|
|
rc
|
The return code.
|
|
data
|
For an open, contains a newly created context.
|
Returns
Nothing
subscriber_notify_cb_t
Purpose
Subscriber callback; called upon receipt of a notification.
Syntax
typedef void (*subscriber_notify_cb_t) (void *arg,
char *event,
char *data,
int datalen);
|
arg
|
Context pointer passed to subscribe (notify_arg).
|
|
event
|
The event reference (URI). The notification event reference matches the subscription, but may contain additional information called event attributes, such as a uid.
|
|
data
|
The body of the notification. A MIME object.
|
|
datalen
|
Length of the data.
|
">
Parameters
|
arg
|
Context pointer passed to subscribe (notify_arg).
|
|
event
|
The event reference (URI). The notification event reference matches the subscription, but may contain additional information called event attributes, such as a uid.
|
|
data
|
The body of the notification. A MIME object.
|
|
datalen
|
Length of the data.
|
Returns
Zero if successful, non-zero otherwise.
subscriber_new_a
Purpose
Creates a new asynchronous subscriber.
Syntax
void subscriber_new_a (pas_dispatcher_t *disp,
void *worker,
const char *host,
unsigned short port,
subscriber_cb_t cbdone,
void *cbarg);
|
disp
|
Thread dispatcher context returned by pas_dispatcher_new.
|
|
worker
|
Application worker. If not NULL, grouped with existing workers created by ENS to service this subscriber session. Used to prevent multiple threads from accessing the subscriber data at the same time. Only usable if the caller creates and dispatches the GDisp context.
|
|
host
|
Notification server host name or IP address.
|
|
port
|
Subscription service port number.
|
|
cbdone
|
The callback invoked when the subscriber session becomes active and subscriptions can be issued.
There are three parameters to cbdone:
cbarg
The first argument.
A status code.
If non-zero, the subscriber could not be created; value specifies cause of the failure.
The new active subscriber (subscriber_t).
|
|
cbarg
|
First argument of cbdone.
|
">
Parameters
|
disp
|
Thread dispatcher context returned by pas_dispatcher_new.
|
|
worker
|
Application worker. If not NULL, grouped with existing workers created by ENS to service this subscriber session. Used to prevent multiple threads from accessing the subscriber data at the same time. Only usable if the caller creates and dispatches the GDisp context.
|
|
host
|
Notification server host name or IP address.
|
|
port
|
Subscription service port number.
|
|
cbdone
|
The callback invoked when the subscriber session becomes active and subscriptions can be issued.
There are three parameters to cbdone:
cbarg
The first argument.
A status code.
If non-zero, the subscriber could not be created; value specifies cause of the failure.
The new active subscriber (subscriber_t).
|
|
cbarg
|
First argument of cbdone.
|
Returns
Nothing. It passes the new active subscriber as third argument of cbdone callback.
subscriber_new_s
Purpose
Creates a new synchronous subscriber.
Syntax
subscriber_t *subscriber_new_s (pas_dispatcher_t *disp,
const char *host,
unsigned short port);
|
disp
|
Publish and subscribe dispatcher returned by pas_dispatcher_new.
|
|
worker
|
Application worker. If not NULL, grouped with existing workers created by ENS to service this publisher session. Used to prevent multiple threads from accessing the publisher data at the same time. Only usable if the caller creates and dispatches the GDisp context.
|
|
host
|
Notification server host name or IP address.
|
|
port
|
Subscription service port number.
|
">
Parameters
|
disp
|
Publish and subscribe dispatcher returned by pas_dispatcher_new.
|
|
worker
|
Application worker. If not NULL, grouped with existing workers created by ENS to service this publisher session. Used to prevent multiple threads from accessing the publisher data at the same time. Only usable if the caller creates and dispatches the GDisp context.
|
|
host
|
Notification server host name or IP address.
|
|
port
|
Subscription service port number.
|
Returns
A new active subscriber (subscriber_t).
subscribe_a
Purpose
Establishes an asynchronous subscription.
Syntax
void subscribe_a (subscriber_t *subscriber,
const char *event_ref,
subscriber_notify_cb_t notify_cb,
void *notify_arg,
subscriber_cb_t cbdone,
void *cbarg):
|
subscriber
|
The subscriber.
|
|
event_ref
|
The event reference. This is a URI identifying the event's source.
|
|
notify_cb
|
The callback invoked upon receipt of a notification matching this subscription.
|
|
notify_arg
|
The first argument of notify_arg. May be called at any time, by any thread, while the subscription is still active.
|
|
cbdone
|
Called when an unsubscribe completes. It has three Parameters:
|
|
cbarg
|
The first argument of cbdone.
|
">
Parameters
|
subscriber
|
The subscriber.
|
|
event_ref
|
The event reference. This is a URI identifying the event's source.
|
|
notify_cb
|
The callback invoked upon receipt of a notification matching this subscription.
|
|
notify_arg
|
The first argument of notify_arg. May be called at any time, by any thread, while the subscription is still active.
|
|
cbdone
|
Called when an unsubscribe completes. It has three Parameters:
|
|
cbarg
|
The first argument of cbdone.
|
Returns
Nothing.
unsubscribe_a
Purpose
Cancels an asynchronous subscription.
Syntax
void unsubscribe_a (subscriber_t *subscriber,
subscription_t *subscription,
subscriber_cb_t cbdone,
void *cbarg);
|
subscriber
|
The disappearing subscriber.
|
|
subscription
|
The subscription to cancel.
|
|
cbdone
|
Called when an unsubscribe completes. It has three parameters:
|
|
cbarg
|
The first argument of cbdone.
|
">
Parameters
|
subscriber
|
The disappearing subscriber.
|
|
subscription
|
The subscription to cancel.
|
|
cbdone
|
Called when an unsubscribe completes. It has three parameters:
|
|
cbarg
|
The first argument of cbdone.
|
Returns
Nothing.
subscriber_delete
Purpose
Terminates a subscriber.
Syntax
void subscriber_delete (subscriber_t *subscriber);
|
subscriber
|
The subscriber to delete.
|
">
Parameters
|
subscriber
|
The subscriber to delete.
|
Returns.
Nothing
subscriber_get_publisher
Purpose
Creates a publisher, using the credentials of the subscriber.
Syntax
struct publisher_struct *subscriber_get_publisher (subscriber_t
*subscriber);
|
subscriber
|
The subscriber whose credentials are used to create the publisher.
|
">
Parameters
|
subscriber
|
The subscriber whose credentials are used to create the publisher.
|
Returns
The publisher, or NULL if creation failed. In case the creation fails, use the publisher_new.
renl_create_subscriber
Purpose
Creates the subscription part of an RENL.
Syntax
renl_t *renl_create_subscriber (subscription_t *subscription,
const char *renl_id,
const char *publisher);
|
subscription
|
The subscription.
|
|
renl_id
|
The unique RENL identifier. This allows two peers to be able to set up multiple RENL's between them.
|
|
publisher
|
The authenticated identity of the peer.
|
">
Parameters
|
subscription
|
The subscription.
|
|
renl_id
|
The unique RENL identifier. This allows two peers to be able to set up multiple RENL's between them.
|
|
publisher
|
The authenticated identity of the peer.
|
Returns
The opaque RENL object.
renl_cancel_subscriber
Purpose
This cancels an RENL. It does not cancel a subscription. It tells ENS not to acknowledge any more notifications received for this subscription. It destroys the RENL object, the application may no longer use this RENL. All RENLs are automatically destroyed when the subscription is canceled. Therefore, this function does not need to be called to free RENL-related memory before deleting a subscriber.
Syntax
void renl_cancel_subscriber (renl_t *renl);
|
renl
|
The RENL to cancel.
|
">
Parameters
Returns
Nothing.
Publish and Subscribe Dispatcher API
The Publish and Subscribe Dispatcher API includes one definition and four functions:
pas_dispatcher_t
Purpose
A publish and subscribe dispatcher.
Syntax
typedef struct pas_dispatcher_struct pas_dispatcher_t;
Parameters
None.
Returns
Nothing.
pas_dispatcher_new
Purpose
Creates or advertises a dispatcher
Syntax
pas_dispatcher_t *pas_dispatcher_new (void *disp);
|
dispcx
|
The dispatcher context. If NULL,to start dispatching notifications, the application must call pas_dispatch.
If not NULL, the dispatcher is a libasync dispatcher.
|
">
Parameters
|
dispcx
|
The dispatcher context. If NULL,to start dispatching notifications, the application must call pas_dispatch.
If not NULL, the dispatcher is a libasync dispatcher.
|
Returns
The dispatcher to use when creating publishers or subscribers (pas_dispatcher_t).
pas_dispatcher_delete
Purpose
Destroys a dispatcher created with pas_dispatcher_new.
Syntax
void pas_dispatcher_delete (pas_dispatcher_t *disp);
|
disp
|
The event notification client environment.
|
">
Parameters
|
disp
|
The event notification client environment.
|
Returns
Nothing.
pas_dispatch
Purpose
Starts the dispatch loop of an event notification environment. It has no effect if the application uses its own thread pool.
Syntax
void pas_dispatch (pas_dispatcher_t *disp);
|
disp
|
The new dispatcher.
|
">
Parameters
Returns
Nothing.
pas_shutdown
Purpose
Stops the dispatch loop of an event notification environment started with pas_dispatch. It has no effect if an application-provided dispatcher was passed to pas_dispatcher_new.
Syntax
void pas_shutdown (pas_dispatcher_t *disp);
|
disp
|
The dispatcher context to shutdown.
|
">
Parameters
|
disp
|
The dispatcher context to shutdown.
|
Returns
Nothing
Sample Code
iPlanet Calendar Server ships with a complete ENS implementation. If you wish to customize it, you may use these APIs to do so. The following four code samples, a simple publisher and subscriber pair, and a reliable publisher and subscriber pair, illustrate how to use the API. The sample code is provided with the product in the following directory:
/opt/SUNWics5/cal/csapi/samples/ens
Simple Publisher and Subscriber
This sample code pair establishes a simple interactive asynchronous publisher and subscriber.
Publisher Code Sample
/*
* Copyright 2000 by Sun Microsystems, Inc.
* All rights reserved
*
* apub : simple interactive asynchronous publisher using
*
* Syntax:
* apub host port
*/
#include <stdlib.h>
#include <stdio.h>
#include "pasdisp.h"
#include "publisher.h"
static pas_dispatcher_t *disp = NULL;
static publisher_t *_publisher = NULL;
static int _shutdown = 0;
static void _read_stdin();
static void _exit_usage()
{
printf("\nUsage:\n apub host port\n");
exit(5);
}
static void _exit_error(const char *msg)
{
printf("%s\n", msg);
exit(1);
}
static void _call_shutdown()
{
_shutdown = 1;
pas_shutdown(disp);
}
static void _open_ack(void *arg, int rc, void *enc)
{
_publisher = (publisher_t *)enc;
(void *)arg;
if (!_publisher)
{
printf("Failed to create publisher with status %d\n", rc);
_call_shutdown();
return;
}
_read_stdin();
return;
}
static void _publish_ack(void *arg, int rc, void *ignored)
{
(void *)ignored;
free(arg);
if (rc != 0)
{
printf("Publish failed with status %d\n", rc);
_call_shutdown();
return;
}
_read_stdin();
return;
}
static void _read_stdin()
{
static char input[1024];
printf("apub> ");
fflush(stdout);
while (!_shutdown)
{
if ( !fgets(input, sizeof(input), stdin) )
{
continue;
} else {
char *message;
unsigned int message_len;
input[strlen(input) - 1] = 0; /* Strip off the \n */
if (*input == '.' && input[1] == 0)
{
publisher_delete(_publisher);
_call_shutdown();
break;
}
message = strdup(input);
message_len = strlen(message);
publish(_publisher, "enp://yoyo.com/xyz",message,
message_len,
_publish_ack, NULL, (void *)message, 0);
return;
}
}
return;
}
main(int argc, char **argv)
{
unsigned short port = 7997;
char host[256];
if (argc < 2) _exit_usage();
if (*(argv[1]) == '0')
{
strcpy(host, "127.0.0.1");
} else {
strcpy(host, argv[1]);
}
if (argc > 2)
{
port = (unsigned short)atoi(argv[2]);
}
disp = pas_dispatcher_new(NULL);
if (disp == NULL) _exit_error("Can't create publisher");
publisher_new_a(disp, NULL, host, port, _open_ack, disp);
pas_dispatch(disp);
_shutdown = 1;
pas_dispatcher_delete(disp);
exit(0);
}
Subscriber Code Sample
/*
* Copyright 1997 by Sun Microsystems, Inc.
* All rights reserved
*
* asub : example asynchronous subscriber
*
* Syntax:
* asub host port
*/
#include <stdlib.h>
#include <stdio.h>
#include "pasdisp.h"
#include "subscriber.h"
static pas_dispatcher_t *disp = NULL;
static subscriber_t *_subscriber = NULL;
static subscription_t *_subscription = NULL;
static renl_t *_renl = NULL;
static void _exit_usage()
{
printf("\nUsage:\n asub host port\n");
exit(5);
}
static void _exit_error(const char *msg)
{
printf("%s\n", msg);
exit(1);
}
static void _subscribe_ack(void *arg, int rc, void *subscription)
{
(void)arg;
if (!rc)
{
_subscription = subscription;
printf("Subscription successful\n");
} else {
printf("Subscription failed - status %d\n", rc);
pas_shutdown(disp);
}
}
static void _unsubscribe_ack(void *arg, int rc, void *ignored)
{
(void *)ignored;
(void *)arg;
if (rc != 0)
{
printf("Unsubscribe failed - status %d\n", rc);
}
subscriber_delete(_subscriber);
pas_shutdown(disp);
}
static int _handle_notify(void *arg, char *url, char *str, int len)
{
(void *)arg;
printf("[%s] %.*s\n", url, len, (str) ? str : "(null)");
return 0;
}
static void _open_ack(void *arg, int rc, void *enc)
{
_subscriber = (subscriber_t *)enc;
(void *)arg;
if (rc)
{
printf("Failed to create subscriber with status %d\n", rc);
pas_shutdown(disp);
return;
}
subscribe(_subscriber, "enp://yoyo.com/xyz",
_handle_notify, NULL,
_subscribe_ack, NULL);
return;
}
static void _unsubscribe(int sig)
{
(int)sig;
unsubscribe(_subscriber, _subscription, _unsubscribe_ack, NULL);
}
main(int argc, char **argv)
{
unsigned short port = 7997;
char host[256];
if (argc < 2) _exit_usage();
if (*(argv[1]) == '0')
{
strcpy(host, "127.0.0.1");
} else {
strcpy(host, argv[1]);
}
if (argc > 2)
{
port = (unsigned short)atoi(argv[2]);
}
disp = pas_dispatcher_new(NULL);
if (disp == NULL) _exit_error("Can't create publisher");
subscriber_new_a(disp, NULL, host, port, _open_ack, NULL);
pas_dispatch(disp);
pas_dispatcher_delete(disp);
exit(0);
}
Reliable Publisher and Subscriber
This sample code pair establishes a reliable asynchronous publisher and subscriber.
Reliable Publisher Sample
/*
* Copyright 2000 by Sun Microsystems, Inc.
* All rights reserved
*
* rpub : simple *reliable* interactive asynchronous publisher.
* It is designed to be used in combination with rsub,
* the reliable subscriber.
*
* Syntax:
* rpub host port
*/
#include <stdlib.h>
#include <stdio.h>
#include "pasdisp.h"
#include "publisher.h"
static pas_dispatcher_t *disp = NULL;
static publisher_t *_publisher = NULL;
static int _shutdown = 0;
static renl_t *_renl;
static void _read_stdin();
static void _exit_usage()
{
printf("\nUsage:\n rpub host port\n");
exit(5);
}
static void _exit_error(const char *msg)
{
printf("%s\n", msg);
exit(1);
}
static void _call_shutdown()
{
_shutdown = 1;
pas_shutdown(disp);
}
static void _renl_create_cb(void *arg, int rc, void *ignored)
{
(void *)arg;
(void *)ignored;
if (!_publisher)
{
printf("Failed to create RENL - status %d\n", rc);
_call_shutdown();
return;
}
_read_stdin();
return;
}
static void _publisher_new_cb(void *arg, int rc, void *enc)
{
_publisher = (publisher_t *)enc;
(void *)arg;
if (!_publisher)
{
printf("Failed to create publisher - status %d\n", rc);
_call_shutdown();
return;
}
renl_create_publisher(_publisher, "renl_id", NULL,
_renl_create_cb,NULL);
return;
}
static void _recv_ack(void *arg, int rc, void *ignored)
{
(void *)ignored;
if (rc < 0)
{
printf("Acknowledgment Timeout\n");
} else if ( rc == 0) {
printf("Acknowledgment Received\n");
}
fflush (stdout);
_read_stdin();
free(arg);
return;
}
static void _read_stdin()
{
static char input[1024];
printf("rpub> ");
fflush(stdout);
while (!_shutdown)
{
if ( !fgets(input, sizeof(input), stdin) )
{
continue;
} else {
char *message;
unsigned int message_len;
input[strlen(input) - 1] = 0; /* Strip off the \n */
if (*input == '.' && input[1] == 0)
{
publisher_delete(_publisher);
_call_shutdown();
break;
}
message = strdup(input);
message_len = strlen(message);
/* five seconds timeout */
publish(_publisher, "enp://yoyo.com/xyz",
message, message_len,
NULL, _recv_ack, message, 5000);
return;
}
}
return;
}
main(int argc, char **argv)
{
unsigned short port = 7997;
char host[256];
if (argc < 2) _exit_usage();
if (*(argv[1]) == '0')
{
strcpy(host, "127.0.0.1");
} else {
strcpy(host, argv[1]);
}
if (argc > 2)
{
port = (unsigned short)atoi(argv[2]);
}
disp = pas_dispatcher_new(NULL);
if (disp == NULL) _exit_error("Can't create publisher");
publisher_new_a(disp, NULL, host, port, _publisher_new_cb,
NULL);
pas_dispatch(disp);
_shutdown = 1;
pas_dispatcher_delete(disp);
exit(0);
}
Reliable Subscriber Sample
/*
* Copyright 1997 by Sun Microsystems, Inc.
* All rights reserved
*
* asub : example asynchronous subscriber
*
* Syntax:
* asub host port
*/
#include <stdlib.h>
#include <stdio.h>
#include "pasdisp.h"
#include "subscriber.h"
static pas_dispatcher_t *disp = NULL;
static subscriber_t *_subscriber = NULL;
static subscription_t *_subscription = NULL;
static renl_t *_renl = NULL;
static void _exit_usage()
{
printf("\nUsage:\n asub host port\n");
exit(5);
}
static void _exit_error(const char *msg)
{
printf("%s\n", msg);
exit(1);
}
static void _subscribe_ack(void *arg, int rc, void *subscription)
{
(void)arg;
if (!rc)
{
_subscription = subscription;
printf("Subscription successful\n");
_renl = renl_create_subscriber(_subscription, "renl_id",
NULL);
} else {
printf("Subscription failed - status %d\n", rc)
pas_shutdown(disp);
}
}
static void _unsubscribe_ack(void *arg, int rc, void *ignored)
{
(void *)ignored;
(void *)arg;
if (rc != 0)
{
printf("Unsubscribe failed - status %d\n", rc);
}
subscriber_delete(_subscriber);
pas_shutdown(disp);
}
static int _handle_notify(void *arg, char *url, char *str, int len)
{
(void *)arg;
printf("[%s] %.*s\n", url, len, (str) ? str : "(null)");
return 0;
}
static void _open_ack(void *arg, int rc, void *enc)
{
_subscriber = (subscriber_t *)enc;
(void *)arg;
if (rc)
{
printf("Failed to create subscriber with status %d\n", rc);
pas_shutdown(disp);
return;
}
subscribe(_subscriber, "enp://yoyo.com/xyz",_handle_notify,
NULL,_subscribe_ack, NULL);
return;
}
static void _unsubscribe(int sig)
{
(int)sig;
unsubscribe(_subscriber, _subscription, _unsubscribe_ack, NULL);
}
main(int argc, char **argv)
{
unsigned short port = 7997;
char host[256];
if (argc < 2) _exit_usage();
if (*(argv[1]) == '0')
{
strcpy(host, "127.0.0.1");
} else {
strcpy(host, argv[1]);
}
if (argc > 2)
{
port = (unsigned short)atoi(argv[2]);
}
disp = pas_dispatcher_new(NULL);
if (disp == NULL) _exit_error("Can't create publisher");
subscriber_new_a(disp, NULL, host, port, _open_ack, NULL);
pas_dispatch(disp);
pas_dispatcher_delete(disp);
exit(0);
}