Watching

Watching information is very different from polling. When watching information, the RTclient that initiated the watch is notified asynchronously, through a message, whenever the item(s) of interest changes. All watching of a project goes through RTserver. Typically, a request to watch particular information is sent to RTserver. RTserver then notifies the process holding the information (the same RTserver, a different RTserver, or an RTclient) that a program is interested in watching the specified information. Then, whenever the information changes, the process holding the information sends it back to the requesting program in a message. Unlike polling, which is a one time request for information, watching causes status messages to continue to be sent until watching is turned off for that particular information.

The type of information watched about projects, RTservers, RTclients, and subjects, is shown in the table. The table lists the entity from which the information is available (such as RTclient, RTserver, or subject), a brief description of the information that is available, whether the information can be polled as well as watched, and the final column shows the TipcMon* API call that initiates the watching and the monitoring message type returned whenever the information of interest changes. The prefix T_MT_ is omitted from the message type for the sake of brevity.

Entity	Information Available	Poll?	API Function and Message Type Returned (T_MT_ omitted)
Project	Names	Yes	TipcMonProjectNamesSetWatch MON_PROJECT_NAMES_STATUS
RTclient	Read and write message buffers	Yes	TipcMonClientBufferSetWatch MON_CLIENT_BUFFER_STATUS
RTclient	Write buffer congestion	No	TipcMonClientCongestionSetWatch MON_CLIENT_CONGESTION_SET_WATCH
RTclient	Messages being received	No	TipcMonClientMsgRecvSetWatch MON_CLIENT_MSG_RECV_STATUS
RTclient	Messages being sent out	No	TipcMonClientMsgSendSetWatch MON_CLIENT_MSG_RECV_STATUS
RTclient	Subjects subscribing to	Yes	TipcMonClientSubscribeSetWatch MON_CLIENT_SUBSCRIBE_STATUS
RTclient	Wall time, CPU time used	Yes	TipcMonClientTimeSetWatch MON_CLIENT_TIME_STATUS
RTclient	Names	Yes	TipcMonClientNamesSetWatch MON_CLIENT_NAMES_STATUS
RTserver	Write buffer congestion	No	TipcMonServerCongestionSetWatch MON_SERVER_CONGESTION_SET_WATCH
RTserver	Connections	Yes	TipcMonServerConnSetWatch MON_SERVER_CONN_STATUS
RTserver	Licenses	No	TipcMonServerMaxClientLicensesSetWatch MON_SERVER_MAX_CLIENT_LICENSES_SET_WATCH
RTserver	Names	Yes	TipcMonServerNamesSetWatch MON_SERVER_NAMES_STATUS
subject	Names	Yes	TipcMonSubjectNamesSetWatch MON_SUBJECT_NAMES_STATUS
subject	Clients subscribing to	Yes	TipcMonSubjectSubscribeSetWatch MON_SUBJECT_SUBSCRIBE_STATUS

Most of the information that can be watched can also be polled. See Polling for more details on polling for information.

To set up watching in your program, you call an API function, TipcMonTypeSetWatch, with the watch_status parameter set to TRUE, where Type is the kind of information to watch. For example, a call to TipcMonSubjectSubscribeSetWatch(subject_name, TRUE) turns on watching for information about which RTclients are subscribing to a specified subject, and a call to TipcMonClientTimeSetWatch(client_name, TRUE) sets up watching for time information about an RTclient.

In a similar fashion, watching is turned off by calling the API function with the watch_status parameter set to FALSE (such as TipcMonSubjectSubscribeSetWatch(subject_name, FALSE)).

Calling a TipcMonTypeSetWatch function with the watch_status parameter set to TRUE causes a message of type MON_TYPE_SET_WATCH to be sent to RTserver, where TYPE is the kind of information being watched. This TYPE is very similar to the Type used in the name of the API function as shown in these examples:

An RTserver may hold the information being watched or it may need to forward the message on to other RTservers or RTclients to watch the information. An example of information that an RTserver keeps track of is a list of subjects to which its direct RTclients are subscribing. An example of the information an RTserver does not know about is the information about an RTclient’s message buffers.

If RTserver does not hold the information and RTclient has the information, RTclient must be reading and processing messages when the initial call to TipcMonTypeSetWatch is made. Messages can be read and processed through calls to TipcSrvMainLoop, TipcSrvMsgNext, TipcMsgSearch, or TipcMsgSearchType when the watching is enabled in order for the watching to take effect. If RTclient does not receive and process a MON_TYPE_STATUS_SET_WATCH message, it will not send out the watched information when it changes.

Whenever the information of interest changes, RTserver sends a message with the information to the program that set up the watch. The watched information is returned by RTserver using a MON_TYPE_STATUS message, where TYPE matches the TYPE in the MON_TYPE_SET_WATCH that initiated the watch. The message types are in pairs of MON_TYPE_SET_WATCH and MON_TYPE_STATUS as shown in these examples:

Processing Watch Results

When the results of a watch are returned in a MON_TYPE_STATUS message, the message can be processed using an RTclient message process callback (created with TipcSrvProcessCbCreate). Using callbacks to process the status messages that are returned as the result of a watch being enabled is the recommended method for processing watch results. The main reason for this is that the messages arrive asynchronously. It is often unclear as to when or how often the information may change. The following section shows an example of how to process watch results using callbacks.

Using Callbacks

This example shows how to use an RTclient message process callback (created with TipcSrvProcessCbCreate) to process the message containing the results of a watch being set up:

/* =============================================================== */ 
/*..process_mon_project_names_status - callback to process a  
                                  MON_PROJECT_NAMES_STATUS message */ 
void T_ENTRY process_mon_project_names_status( 
                      T_IPC_CONN conn, 
                      T_IPC_CONN_PROCESS_CB_DATA data, 
                      T_CB_ARG arg) 
{ 
  T_STR *project_names; 
  T_INT4 num_project_names; 
  T_STR created_project; 
  T_STR destroyed_project; 
  T_INT4 counter; 
 
  /* access fields of status message */ 
  if (!TipcMsgSetCurrent(data->msg, 0)) { 
    /* error */ 
  }  
 
  if (!TipcMsgNextStrArray(data->msg, &project_names,  
                           &num_project_names) 
      || !TipcMsgNextStr(data->msg, &created_project) 
      || !TipcMsgNextStr(data->msg, &destroyed_project)) { 
    /* error */ 
  } 
 
  /* ...do whatever with results here */ 
 
} /* process_mon_project_names_status */ 
 
/* =========================================================== */ 
/*...code from calling function is below */ 
 
T_IPC_MT mt; 
 
/* setup watching of projects */ 
if (!TipcMonProjectNamesSetWatch(TRUE)) { 
  /* error */ 
}  
 
/* create callback to process MON_PROJECT_NAMES_STATUS messages */ 
mt = TipcMtLookupByNum(T_MT_MON_PROJECT_NAMES_STATUS); 
if (mt == NULL) { 
  /* error */ 
} 
 
if (TipcSrvProcessCbCreate(mt, process_mon_project_names_status, 
NULL) 
    == NULL) { 
  /* error */ 
} 
 
/* At this point TipcSrvMainLoop can be used to read and process messages. */

Printing Watch Categories

The API function TipcMonPrintWatch can be used to print all the monitoring categories being watched by a program to a TutOut-style output function. The output function is called once for each line of output as shown in the example:

/* Set up some information to be watched */ 
if (!TipcMonSubjectSubscribeSetWatch(T_IPC_MON_ALL, TRUE) { 
  /* error */ 
}  
if (!TipcMonProjectNamesSetWatch(TRUE) { 
  /* error */ 
}  
if (!TipcMonClientTimeSetWatch("primary_rtie", TRUE) { 
  /* error */ 
}  
if (!TipcMonClientMsgRecvSetWatch("primary_rtie", "numeric_data", 
                                  TRUE) { 
  /* error */ 
} 
if (!TipcMonPrintWatch(TutOut)) { 
  /* error */ 
}

Watching project_names. 
Watching subject_subscribe <*>. 
Watching client_time <primary_rtie>. 
Watching client_msg_recv <primary_rtie> <numeric_data>.

Watching Message Types

This section contains a complete listing of each watching message type. For each message pair (MON_TYPE_SET_WATCH and MON_TYPE_STATUS), a description of the type of information, the API function, and where the information is collected from is shown.

The T_MT_ prefix is omitted from the name of each message type for the sake of brevity.

MON_CLIENT_BUFFER_*

The MON_CLIENT_BUFFER_SET_WATCH message type is used to set whether or not an RTclient is watching message-related buffer information in one or more RTclients in a project. When watching an RTclient’s message buffers, RTserver sends a MON_CLIENT_BUFFER_STATUS message each time the message queue for the connection to RTserver changes in the watched RTclients. When watching is first enabled, an initial status message is sent if the RTclient(s) being watched exists.

MON_CLIENT_CONGESTION_*

The MON_CLIENT_CONGESTION_SET_WATCH message type is used to set whether or not an RTclient is watching for congestion in another RTclient’s write buffer. When watching an RTclient’s write buffers, RTserver sends a MON_CLIENT_CONGESTION_STATUS message if the number of pending messages in the write buffer reaches a set amount. Another message is sent when the number of pending messages decreases to a lower set amount.

MON_CLIENT_MSG_RECV_*

The MON_CLIENT_MSG_RECV_SET_WATCH message type is used to set whether or not an RTclient is watching message-received information in one or more RTclients in a project.When watching an RTclient’s incoming messages, RTserver sends a MON_CLIENT_MSG_RECV_STATUS message each time a received message is inserted into or deleted from the message queue (for the connection to RTserver) in the watched RTclients. When watching is first enabled, an initial status message is sent if the RTclient(s) being watched exists.

MON_CLIENT_MSG_SEND_*

The MON_CLIENT_MSG_SEND_SET_WATCH message type is used to set whether or not an RTclient is watching sent messages in one or more RTclients in a project. When watching an RTclient’s outgoing messages, RTserver sends a MON_CLIENT_MSG_SEND_STATUS message each time a message is sent to RTserver from the watched RTclients. When watching is first enabled, an initial status message is sent if the RTclient(s) being watched exists.

MON_CLIENT_NAMES_*

The MON_CLIENT_NAMES_SET_WATCH message type is used to set whether or not an RTclient is watching RTclient names in a project. When watching RTclient names, RTserver sends a MON_CLIENT_NAMES_STATUS message each time an RTclient is created or destroyed. When watching is first enabled, an initial status message is sent if an RTclient is running. An RTclient is considered created when it connects to RTserver. An RTclient is considered destroyed when when its connection to RTserver is closed or lost.

MON_CLIENT_SUBSCRIBE_*

The MON_CLIENT_SUBSCRIBE_SET_WATCH message type is used to set whether or not an RTclient is watching the subjects to which one or more RTclients in a project are subscribing. When watching RTclient subscriptions, RTserver sends a MON_CLIENT_SUBSCRIBE_STATUS message each time the watched RTclients start or stop subscribing to a subject. When watching is first enabled, an initial status message is sent if the RTclient(s) being watched exists.

MON_CLIENT_TIME_*

The MON_CLIENT_TIME_SET_WATCH message type is used to set whether or not an RTclient is watching time information in one or more RTclients in a project. When watching an RTclient’s time information, RTserver sends a MON_CLIENT_TIME_STATUS message each time the current time changes in the watched RTclients. When watching is first enabled, an initial status message is sent if the RTclient(s) being watched exists.

MON_PROJECT_NAMES_*

The MON_PROJECT_NAMES_SET_WATCH message type is used to set whether or not an RTclient is watching project names. When watching project names, RTserver sends a MON_PROJECT_NAMES_STATUS message each time a project is created or destroyed, as well as an initial status message when watching is first enabled.

MON_SERVER_CONGESTION_*

The MON_SERVER_CONGESTION_SET_WATCH message type is used to set whether or not an RTclient is watching for congestion in a process’s write buffer. When watching the process’s write buffers, RTserver sends a MON_SERVER_CONGESTION_STATUS message if the number of pending bytes in the write buffer reaches a set amount. Another message is sent when the number of pending bytes decreases to a lower set amount.

MON_SERVER_CONN_*

The MON_SERVER_CONN_SET_WATCH message type is used to set whether or not an RTclient is watching RTserver connections. When watching RTserver connections, RTserver sends a MON_SERVER_CONN_STATUS message each time a connection between RTservers is created or destroyed.

MON_SERVER_MAX_CLIENT_LICENSES_*

The MON_SERVER_MAX_CLIENT_LICENSES_SET_WATCH message type is used to set whether or not an RTserver is watching for the number of client connections to reach and exceeded the licensed amount. When watching client connection licenses, RTserver sends a MON_SERVER_MAX_CLIENT_LICENSES_STATUS message each time an RTserver refuses a client connection because no licenses were available.

MON_SERVER_NAMES_*

The MON_SERVER_NAMES_SET_WATCH message type is used to set whether or not an RTclient is watching RTserver names. When watching RTserver names, RTserver sends a MON_SERVER_NAMES_STATUS message each time an RTserver is created or destroyed. An RTserver is considered created when it starts up (and usually connects to other RTservers). An RTserver is considered destroyed when it terminates or disconnects from other RTservers.

MON_SUBJECT_NAMES_*

The MON_SUBJECT_NAMES_SET_WATCH message type is used to set whether or not an RTclient is watching subject names in a project. When watching subject names, RTserver sends a MON_SUBJECT_NAMES_STATUS message each time a subject is created or destroyed in the project. A subject is considered created when the first RTclient starts subscribing to that subject. A subject is considered destroyed when the last RTclient stops subscribing to that subject.

MON_SUBJECT_SUBSCRIBE_*

The MON_SUBJECT_SUBSCRIBE_SET_WATCH message type is used to set whether or not an RTclient is watching the RTclients that are subscribing to a specified subject or any subject. When watching subjects being subscribed to, RTserver sends a MON_SUBJECT_SUBSCRIBE_STATUS message each time an RTclient starts or stops subscribing to the subject(s). When watching is first enabled, an initial status message is sent if the subject being watched exists.

Watching Example

The following is a complete C source example that watches a subject and outputs information whenever an RTclient starts subscribing to a subject or stops subscribing to a subject.

$RTHOME/examples/smrtsock/manual

RTHOME:[EXAMPLES.SMRTSOCK.MANUAL]

%RTHOME%\examples\smrtsock\manual

The online source files have additional #ifdefs to provide C++ support; these #ifdefs are not shown to simplify the example.

/* 
  -------------------------------------------------------------------- 
  watchsbj.c                                                           
                                                                       
  USAGE                                                               
    watchsbj <project> <subject>                                
 
  This program connects to RTserver and outputs a count and names     
  of RTclients that subscribed to <subject> in <project>.    
  Whenever an RTclient subscribes or unsubscribes to a subject,  
   information about the subject is output.                                       
  -------------------------------------------------------------------  
*/ 
 
#include <rtworks/ipc.h> 
 
/* =============================================================== */ 
/*..cb_default - callback to process anything but  
                        MON_SUBJECT_SUBSCRIBE_STATUS */ 
static void T_ENTRY cb_default( 
                      T_IPC_CONN conn, 
                      T_IPC_CONN_PROCESS_CB_DATA data, 
                      T_CB_ARG arg) 
{ 
  TipcMsgPrintError(data->msg); 
} /* cb_default */ 
 
/* =============================================================== */ 
/*..process_subject_status - callback to process MON_SUBJECT_SUBSCRIBE_STATUS msg */ 
static void T_ENTRY process_subject_status( 
                      T_IPC_CONN conn, 
                      T_IPC_CONN_PROCESS_CB_DATA data, 
                      T_CB_ARG arg) 
{ 
  T_IPC_MSG msg = data->msg; 
  T_STR subject_name; 
  T_STR *subscribe_client_names; 
  T_INT4 num_subscribe_clients; 
  T_STR start_client; 
  T_STR stop_client; 
  T_INT4 n;   
 
  /* Set current field */ 
  if (!TipcMsgSetCurrent(msg, 0)) { 
    TutOut("Could not set current field of message: error <%s>.\n", 
           TutErrStrGet()); 
    return; 
  } 
 
  /* Get the fields from the message */ 
  if (!TipcMsgNextStr(msg, &subject_name) 
      || !TipcMsgNextStrArray(msg, &subscribe_client_names,  
                              &num_subscribe_clients) 
      || !TipcMsgNextStr(msg, &start_client) 
      || !TipcMsgNextStr(msg, &stop_client)) { 
    TutOut("Unable to access MON_SUBJECT_SUBSCRIBE_STATUS"); 
    TutOut(" message: error <%s>.\n", 
           TutErrStrGet()); 
    return; 
  } 
 
  /* Print out the new information extracted from message */ 
  TutOut("Received change notice for subject <%s>\n", 
subject_name); 
  TutOut("Number of clients subscribed to <%s> = %d\n",  
         subject_name, num_subscribe_clients); 
  for (n = 0; n < num_subscribe_clients; n++) { 
    TutOut("   [%d] %s\n", n, subscribe_client_names[n]); 
  } 
 
  TutOut("RTclient who just subscribed:   %s\n", start_client); 
  TutOut("RTclient who just unsubscribed: %s\n", stop_client); 
  
TutOut("======================================================\n")
; 
   
} /* process_subject_status */ 
 
/* =============================================================== */ 
int main(argc, argv) 
int argc; 
char **argv; 
{ 
  T_STR project_name; 
  T_STR subject_name; 
  T_IPC_MT mt; 
  T_OPTION option; 
 
  /* Check the command line arguments */ 
  if (argc != 3) { 
    TutOut("Usage: watchsbj <project> <subject>\n"); 
    TutExit(T_EXIT_FAILURE); 
  }  
 
  /* Save pointers to command line arguments */ 
  project_name = argv[1]; 
  subject_name = argv[2]; 
 
  TutOut("Watching subject <%s> in project <%s>...\n", 
         subject_name, project_name); 
 
  /* Set the project name */ 
  option = TutOptionLookup("project"); 
  if (option == NULL) { 
    TutOut("Could not look up option <project>: error <%s>.\n", 
           TutErrStrGet()); 
    TutExit(T_EXIT_FAILURE); 
  }  
 
  if (!TutOptionSetEnum(option, project_name)) { 
    TutOut("Could not set option <project>: error <%s>.\n", 
           TutErrStrGet()); 
    TutExit(T_EXIT_FAILURE); 
  }  
 
  /* Connect to RTserver */ 
  if (!TipcSrvCreate(T_IPC_SRV_CONN_FULL)) { 
    TutOut("Could not connect to RTserver: error <%s>.\n", 
           TutErrStrGet()); 
    TutExit(T_EXIT_FAILURE); 
  }  
 
  /* Create callback to process subject status messages */ 
  mt = TipcMtLookupByNum(T_MT_MON_SUBJECT_SUBSCRIBE_STATUS); 
  if (mt == NULL) { 
    TutOut("Could not look up message type");  
    TutOut(" MON_SUBJECT_SUBSCRIBE_STATUS: error <%s>.\n", 
           TutErrStrGet()); 
    TutExit(T_EXIT_FAILURE); 
  }  
 
  if (TipcSrvProcessCbCreate(mt, process_subject_status, NULL) 
      == NULL) { 
    TutOut("Could not create callback to process "); 
    TutOut("MON_SUBJECT_SUBSCRIBE_STATUS: error <%s>.\n", 
           TutErrStrGet()); 
    TutExit(T_EXIT_FAILURE); 
  }  
 
  /* Create default callback to receive unwanted message types */ 
  if (TipcSrvDefaultCbCreate(cb_default, NULL) == NULL) { 
    TutOut("Could not create default callback: error <%s>.\n", 
           TutErrStrGet()); 
  } 
 
  /* Start "watching" the subject */ 
  if (!TipcMonSubjectSubscribeSetWatch(subject_name, TRUE)) { 
    TutOut("Could not start watching "); 
    TutOut("subject <%s>: error <%s>.\n", 
           subject_name, TutErrStrGet()); 
    TutExit(T_EXIT_FAILURE); 
  }  
 
  /* If an error occurs, then TipcSrvMainLoop will restart RTserver */ 
  /* and return FALSE. We can safely continue. */ 
  for (;;) { 
    if (!TipcSrvMainLoop(T_TIMEOUT_FOREVER)) { 
      TutOut("TipcSrvMainLoop failed: error <%s>.\n",  
             TutErrStrGet()); 
    }  
  } 
 
  /* This line should not be reached. */ 
  return T_EXIT_FAILURE; 
} /* main */

The call to TipcMonSubjectSubscribeSetWatch turns on watching for all RTclients subscribing to the specified subject. Whenever an RTclient starts or stops subscribing to the specified subject, RTserver asynchronously sends a MON_SUBJECT_SUBSCRIBE_STATUS message to the watchsbj program. Prior to turning watching on, a callback (process_subject_status) is created to process a message of this type when it arrives. The callback function process_subject_status then prints out the information contained in the message.

Initially, if there are no RTclients subscribing to the specified subject, no message is sent by RTserver to the program.

Compiling, Linking, and Running

To compile, link, and run the watchsbj program, first you must either copy the program to your own directory or have write permission in these directories:

$RTHOME/examples/smrtsock/manual

RTHOME:[EXAMPLES.SMRTSOCK.MANUAL]

%RTHOME%\examples\smrtsock\manual

Step 1

Note that this program is part of the ss_tutorial project and subscribes to the lesson5 subject.

$RTHOME/examples/smrtsock/tutorial/lesson5

RTHOME:[EXAMPLES.SMRTSOCK.TUTORIAL.LESSON5]

%RTHOME%\examples\smrtsock\tutorial\lesson5

Step 2

$ rtlink watchsbj.c -o watchsbj.x 
$ rtlink receive.c -o receive.x

$ cc watchsbj.c 
$ rtlink /exec=watchsbj.exe watchsbj.obj 
$ cc receive.c 
$ rtlink /exec=receive.exe receive.obj

$ nmake /f wsjw32m.mak 
$ nmake /f recw32m.mak

To run the programs, start the receive process in one terminal emulator window and then the watchsbj process in another terminal emulator window.

Step 3

Step 4

$ watchsbj.x ss_tutorial lesson5

$ watchsbj :== $sys$disk:[]watchsbj.exe 
$ watchsbj ss_tutorial lesson5

$ watchsbj.exe ss_tutorial lesson5

Watching subject <lesson5> in project <ss_tutorial>... 
Connecting to project <ss_tutorial> on <_node> RTserver. 
Using local protocol. 
Message from RTserver: Connection established. 
Start subscribing to subject </_grissom_1026>. 
Received change notice for subject <lesson5> 
/* This is the initial status message which is returned */ 
Number of clients subscribed to <lesson5> = 1 
   [0] /_grissom_1018 
RTclient who just subscribed:    
RTclient who just unsubscribed:  
======================================================

Step 5

Received change notice for subject <lesson5> 
Number of clients subscribed to <lesson5> = 0 
RTclient who just subscribed:    
RTclient who just unsubscribed: /_grissom_1018 
====================================================== 
Received change notice for subject <lesson5> 
Number of clients subscribed to <lesson5> = 1 
   [0] /_grissom_1028 
RTclient who just subscribed:   /_grissom_1028 
RTclient who just unsubscribed:  
======================================================

Note that watchsbj output results immediately when the program was killed and again when it was restarted.