Galaxy Communicator Documentation:

Upgrading from 2.1 to 3.0

License / Documentation home / Help and feedback

The 3.0 Galaxy Communicator distribution contains major improvements to outgoing brokering. There are obligatory upgrades associated with these improvements; in particular, APIs have unavoidably changed. In almost all other ways, the 3.0 upgrade should be transparent for C programmers; except for outgoing brokering, the programmer will probably not have exploited the functionality which needs to be changed.  For Java programmers, a significant internal reorganization of the Java bindings was required to bring its functionality in line with the C library, and there are a number of unavoidable API changes which are required. For Python and Allegro programmers, the changes are much more modest, but in many cases still necessary.

This document describes only the steps required or recommended to upgrade existing features of Galaxy Communicator. You can find a list of new features here. You can find the complete release notes here.
 
 

Step	Who's affected	Status
Step 1: Configuring your system	Users of the Python and Allegro bindings, users of the MITRE PostGres DB wrapper server	Required
Step 2a: Upgrading outgoing brokering	All server developers	Required
Step 2b: Upgrading memory management in broker callbacks	Users of incoming brokering in the libGalaxy C library	Optional, but highly recommended for efficient performance
Step 2c: Upgrading incoming brokering	Users of incoming brokering in the libGalaxy C library	Optional
Step 2d: Using broker callbacks for completion and aborting	Users of brokering	Optional
Step 3: Upgrading Python bindings	Users of the Python bindings	Required
Step 4: Upgrading Java bindings	Users of the Java bindings	Required
Step 5: Upgrading server listen status	Users of the  server listen status functionality in the libGalaxy C library (introduced in 2.1)	Required
Step 6: Upgrading references to the call environment	Users of the libGalaxy C library who inadvisedly exploited the odd return properties  in 2.1 of GalSS_EnvError and GalSS_EnvDestroyToken	Required
Step 7: Upgrading calls to GalIO_SetServerData	Users of the libGalaxy C library who use either use GalIO_SetServerData to store arbitrary data in the server, or returned values from _GalSS_init_server for storage	Required (GalIO_SetServerData), recommended (_GalSS_init_server)
Step 8a: Upgrading references to the Communicator main loop	Users of the libGalaxy C library who have written servers with their own main()	Optional
Step 8b: Upgrade your own main loop	Users of the libGalaxy C library who have written servers with their own event loops (embedded in Tcl/Tk or CORBA, for instance)	Optional but highly recommended
Step 9: Upgrade timed task invocation	Users of the libGalaxy C library who set up their own timed tasks	Optional
Step 10: Upgrading session management	Users of the libGalaxy C library	Optional but highly recommended, especially for those who intend to run multiple parallel sessions

Even though a number of these steps are optional, we strongly encourage you to consider them. The functionality they replace has not been deprecated, but in many cases, future upgrades will not extend the older functionality. In addition, steps 8-10 feature radically improved strategies for embedding Communicator servers in external main loops, and for session handling; while these upgrades are optional, these improvements are so dramatic that we strongly recommend that everyone adopt them.

Step 1: Configuring your system

In addition, the 3.0 Galaxy Communicator distribution requires Allegro 5.0.

Finally, we have added the PostGres DB server to the default compilation. If you want to enable this server, uncomment the POSTGRES_INCLUDE and POSTGRES_LIB variables in your config file and set them to the appropriate values.

Step 2a: Updating outgoing brokering

Here's the old way of doing outgoing brokering:

static Gal_Frame prepare_audio_frame(char *filename)
{
  Gal_Frame f = Gal_MakeFrame("main", GAL_CLAUSE);
  int total;
  char *buf;
  GalIO_BrokerStruct *b;
  char host_pid[1024];

  /* In the code omitted here, the data from the named file is
     read into buf, and total is the number of bytes in buf */

  /* .... */

  /* Now that we have the audio, we add a binary element. */
  /* These must be added to the broker frame BEFORE the
     broker object is initialized. */
  sprintf(host_pid, "%s:%d", GalIO_IPAddress(), (int) getpid());
  GalIO_FrameSetBrokerCallID(f, Gal_StringObject(host_pid));
  b = GalIO_BrokerDataOutInit(3900, f, 0, 0);
  if (b && (GalIO_FrameGetBrokerPort(f) > 0)) {
    Gal_SetProp(f, ":binary_host", Gal_StringObject(GalIO_IPAddress()));
    Gal_SetProp(f, ":binary_port",
                Gal_IntObject(GalIO_FrameGetBrokerPort(f)));
    GalIO_BrokerWriteBinary(b, buf, total);
    GalIO_BrokerDataOutDone(b);
  }
  return f;
}

static Gal_Frame prepare_audio_frame(GalIO_CommStruct *gcomm, char *filename)
{
  Gal_Frame f = Gal_MakeFrame("main", GAL_CLAUSE);
  int total;
  char *buf;
  GalIO_BrokerStruct *b;
  char host_pid[1024];

  /* In the code omitted here, the data from the named file is
     read into buf, and total is the number of bytes in buf */

  /* .... */

  /* Now that we have the audio, we write the binary data
     through the broker. */

  b = GalIO_BrokerDataOutInit(gcomm, 0, 10);
  if (b && (GalIO_GetBrokerListenPort(b) > 0)) {
    GalIO_BrokerPopulateFrame(b, f, ":binary_host", ":binary_port");
    GalIO_BrokerWriteBinary(b, buf, total);
    GalIO_BrokerDataOutDone(b);
  }
  free(buf);
  return f;
}

• It is no longer necessary to create the unique call ID; the broker does this for you.
• GalIO_BrokerDataOutInit no longer takes a port number as its first argument; instead, it takes a connection object, through which it can associate itself with an existing server listener (or start up a listener for the server).
• The third argument to GalIO_BrokerDataOutInit is still a timeout, but it is much more important to set this timeout appropriately, because the semantics of the timeout have changed. Previously, when brokers accepted only a single connection, the broker would destroy itself after it received and satisfied that single connection, and the timeout was only activated if no connection was ever received. Now, since brokers can accept multiple connections, the broker will not be destroyed until the timeout is reached and all currently connected broker clients have received all their data. A broker which never expires will never be freed, and all the data it is intended to send will also remain unfreed, since the broker must cache its data for subsequent connections.
• The host and port for the broker connection are now inserted by GalIO_BrokerPopulateFrame.

Corresponding modifications are required in Allegro, Java and Python.

Step 2b: Upgrading memory management in broker callbacks

Version 2.1

static void audio_handler(GalIO_BrokerStruct *broker_struct,
                          void *data, Gal_ObjectType data_type,
                          int n_samples)
{
  DataHandler *d = (DataHandler *) GalIO_GetBrokerCallerData(broker_struct);

  switch (data_type) {
  case GAL_BINARY:
    if (d->data_buf)
      d->data_buf = (char *) realloc(d->data_buf, n_samples + d->size);
    else
      d->data_buf = (char *) malloc(n_samples + d->size);
    bcopy(data, d->data_buf + d->size, n_samples);
    d->size += n_samples;
    break;
  default:
    GalUtil_Warn("Unknown data type %s\n", Gal_ObjectTypeString(data_type));
  }
}

static void audio_handler(GalIO_BrokerStruct *broker_struct,
                          void *data, Gal_ObjectType data_type,
                          int n_samples)
{
  DataHandler *d = (DataHandler *) GalIO_GetBrokerCallerData(broker_struct);

  switch (data_type) {
  case GAL_BINARY:
    if (d->data_buf)
      d->data_buf = (char *) realloc(d->data_buf, n_samples + d->size);
    else
      d->data_buf = (char *) malloc(n_samples + d->size);
    bcopy(data, d->data_buf + d->size, n_samples);
    d->size += n_samples;
    free(data);
    break;
  default:
    GalUtil_Warn("Unknown data type %s\n", Gal_ObjectTypeString(data_type));
  }
}

Step 2c: Upgrading incoming brokering (optional)

Version 2.x

Gal_Frame receive_audio(Gal_Frame f, void *server_data)
{
  DataHandler *d = (DataHandler *) malloc(sizeof(DataHandler));
  GalIO_BrokerStruct *b;
  char *host = Gal_GetString(f, ":binary_host");
  int port = Gal_GetInt(f, ":binary_port");

  d->data_buf = (char *) NULL;
  d->size = 0;

  if (host && port) {
    b = GalIO_BrokerDataInInit(host, port, f, audio_handler, d, 0);
    if (b) {
      GalIO_SetBrokerActive(b);
    }
  } else {
    free(d);
  }
  return (Gal_Frame) NULL;
}

Gal_Frame receive_audio(Gal_Frame f, void *server_data)
{
  DataHandler *d = (DataHandler *) malloc(sizeof(DataHandler));
  GalIO_BrokerStruct *b;
  char *host = Gal_GetString(f, ":binary_host");
  int port = Gal_GetInt(f, ":binary_port");

  d->data_buf = (char *) NULL;
  d->size = 0;
  d->gcomm = GalSS_EnvComm((GalSS_Environment *) server_data);

  if (host && port) {
     b = GalIO_CommBrokerDataInInit(d->gcomm, host, port, f, audio_handler,
                                    0, d, __FreeDataHandler);
    if (b) {
      GalIO_SetBrokerActive(b);
    }
  } else {
    free(d);
  }
  return (Gal_Frame) NULL;
}

Second, we now discourage the user of GalIO_BrokerSetFinalizer, which has been superseded by the event-driven programming model: Note that this example also illustrates the upgrade to GalSS_EnvBrokerDataInInit, but we won't highlight that contrast here:

Version 2.x

void __AudioSupport_RetrieveAudio(char *host, int port,
                                  int sample_rate,
                                  Gal_Frame f,
                                GalIO_BrokerDataFinalizer finalizer,
                                  void *callback_data)
{
  GalIO_BrokerStruct *b;

  if (host && port && (sample_rate > 0)) {
    DataContainer *c = (DataContainer *) calloc(1, sizeof(DataContainer));
    c->data_buf = (char *) NULL;
    c->size = 0;
    c->callback_data = callback_data;
    b = GalIO_BrokerDataInInit(host, port, f, __AudioCallback, c, 0);

    if (b) {

      /* Set the broker finalizer. */
      GalIO_BrokerSetFinalizer(b, finalizer);

      /* Make the broker active. Without this step, nothing will happen.
         This utility can arguably be used to manage multiple incoming
         audio signals, but it's probably inadequate. */
      GalIO_SetBrokerActive(b);
    }
  }
}

void __AudioSupport_RetrieveAudio(GalSS_Environment *env,
                                  char *host, int port,
                                  int sample_rate,
                                  Gal_Frame f,
                                GalIO_BrokerCallbackFn finalizer,
                                  void *callback_data)
{
  GalIO_BrokerStruct *b;

  if (host && port && (sample_rate > 0)) {
    DataContainer *c = (DataContainer *) calloc(1, sizeof(DataContainer));
    c->data_buf = (char *) NULL;
    c->size = 0;
    c->callback_data = callback_data;
    b = GalSS_EnvBrokerDataInInit(env, host, port, f, __AudioCallback,
                                  0, (void *) c, NULL);

    if (b) {

      GalIO_AddBrokerCallback(b, GAL_BROKER_DATA_DONE_EVENT,
                              finalizer, (void *) NULL);

      /* Make the broker active. Without this step, nothing will happen.
         This utility can arguably be used to manage multiple incoming
         audio signals, but it's probably inadequate. */
      GalIO_SetBrokerActive(b);
    }
  }
}

Step 2d: Using broker callbacks for completion and aborting

With semaphores

#define AUDIO_END "end"
#define AUDIO_ABORT "abort"

static void audio_handler(GalIO_BrokerStruct *broker_struct,
                          void *data, Gal_ObjectType data_type,
                          int n_samples)
{
  DataHandler *d = (DataHandler *) GalIO_GetBrokerData(broker_struct);

  switch (data_type) {
  case GAL_STRING:
    if (!strcmp((char *) data, AUDIO_END)) {
      GalIO_BrokerDataDone(broker_struct);
      __report_done(broker_struct, (void *) NULL);
    } else if (!strcmp((char *) data, AUDIO_ABORT) {
      __report_abort(broker_struct, (void *) NULL);
    } else {
      GalUtil_Warn("Unknown message %s\n", (char *) data);
    }
    free(data);
    break;
  case GAL_BINARY:
    if (d->data_buf)
      d->data_buf = (char *) realloc(d->data_buf, n_samples + d->size);
    else
      d->data_buf = (char *) malloc(n_samples + d->size);
    bcopy(data, d->data_buf + d->size, n_samples);
    d->size += n_samples;
    free(data);
    break;
  default:
    GalUtil_Warn("Unknown data type %s\n", Gal_ObjectTypeString(data_type));
  }
}

void __FreeDataHandler(void *data)
{
  DataHandler *d = (DataHandler *) data;
  free(d->data_buf);
  free(d);
}

void __report_abort(GalIO_BrokerStruct *b, void *data)
{
  DataHandler *d = (DataHandler *) GalIO_GetBrokerData(b);
  Gal_Frame f;

  f = Gal_MakeFrame("notify", GAL_CLAUSE);
  Gal_SetProp(f, ":notification", Gal_StringObject("Audio aborted."));
  GalIO_CommWriteFrame(d->gcomm, f, 0);
  Gal_FreeFrame(f);
}

void __report_done(GalIO_BrokerStruct *b, void *data)
{
  DataHandler *d = (DataHandler *) GalIO_GetBrokerData(b);
  Gal_Frame f;
  FILE *fp = fopen("/dev/audio", "w");

  if (!fp) {
    GalUtil_Warn("Couldn't open /dev/audio");
  } else {
    fwrite(d->data_buf, sizeof(char), d->size, fp);
    fflush(fp);
    fclose(fp);
  }

  f = Gal_MakeFrame("notify", GAL_CLAUSE);
  Gal_SetProp(f, ":notification", Gal_StringObject("Audio received."));
  GalIO_CommWriteFrame(d->gcomm, f, 0);
  Gal_FreeFrame(f);
}

Gal_Frame receive_audio(Gal_Frame f, void *server_data)
{
  DataHandler *d = (DataHandler *) malloc(sizeof(DataHandler));
  GalIO_BrokerStruct *b;
  char *host = Gal_GetString(f, ":binary_host");
  int port = Gal_GetInt(f, ":binary_port");

  d->data_buf = (char *) NULL;
  d->size = 0;
  d->gcomm = GalSS_EnvComm((GalSS_Environment *) server_data);

  if (host && port) {
     b = GalIO_CommBrokerDataInInit(d->gcomm, host, port, f, audio_handler,
                                    0, d, __FreeDataHandler);
    if (b) {
      GalIO_SetBrokerActive(b);
    }
  } else {
    free(d);
  }
  return (Gal_Frame) NULL;
}

Without semaphores

static void audio_handler(GalIO_BrokerStruct *broker_struct,
                          void *data, Gal_ObjectType data_type,
                          int n_samples)
{
  DataHandler *d = (DataHandler *) GalIO_GetBrokerData(broker_struct);

  switch (data_type) {
  case GAL_BINARY:
    if (d->data_buf)
      d->data_buf = (char *) realloc(d->data_buf, n_samples + d->size);
    else
      d->data_buf = (char *) malloc(n_samples + d->size);
    bcopy(data, d->data_buf + d->size, n_samples);
    d->size += n_samples;
    free(data);
    break;
  default:
    GalUtil_Warn("Unknown data type %s\n", Gal_ObjectTypeString(data_type));
  }
}

void __FreeDataHandler(void *data)
{
  DataHandler *d = (DataHandler *) data;
  free(d->data_buf);
  free(d);
}

void __report_abort(GalIO_BrokerStruct *b, void *data)
{
  DataHandler *d = (DataHandler *) GalIO_GetBrokerData(b);
  Gal_Frame f;

  f = Gal_MakeFrame("notify", GAL_CLAUSE);
  Gal_SetProp(f, ":notification", Gal_StringObject("Audio aborted."));
  GalIO_CommWriteFrame(d->gcomm, f, 0);
  Gal_FreeFrame(f);
}

void __report_done(GalIO_BrokerStruct *b, void *data)
{
  DataHandler *d = (DataHandler *) GalIO_GetBrokerData(b);
  Gal_Frame f;
  FILE *fp = fopen("/dev/audio", "w");

  if (!fp) {
    GalUtil_Warn("Couldn't open /dev/audio");
  } else {
    fwrite(d->data_buf, sizeof(char), d->size, fp);
    fflush(fp);
    fclose(fp);
  }

  f = Gal_MakeFrame("notify", GAL_CLAUSE);
  Gal_SetProp(f, ":notification", Gal_StringObject("Audio received."));
  GalIO_CommWriteFrame(d->gcomm, f, 0);
  Gal_FreeFrame(f);
}

Gal_Frame receive_audio(Gal_Frame f, void *server_data)
{
  DataHandler *d = (DataHandler *) malloc(sizeof(DataHandler));
  GalIO_BrokerStruct *b;
  char *host = Gal_GetString(f, ":binary_host");
  int port = Gal_GetInt(f, ":binary_port");

  d->data_buf = (char *) NULL;
  d->size = 0;
  d->gcomm = GalSS_EnvComm((GalSS_Environment *) server_data);

  if (host && port) {
     b = GalIO_CommBrokerDataInInit(d->gcomm, host, port, f, audio_handler,
                                    0, d, __FreeDataHandler);
    if (b) {
      GalIO_SetBrokerActive(b);
      GalIO_AddBrokerCallback(b, GAL_BROKER_ABORT_EVENT,
         __report_abort, (void *) NULL);
      GalIO_AddBrokerCallback(b, GAL_BROKER_DATA_DONE_EVENT,
         __report_done, (void *) NULL);
    }
  } else {
    free(d);
  }
  return (Gal_Frame) NULL;
}

Step 3: Upgrading Python bindings

First, because Python now uses the C library, a number of the elements you could set and access via instance attributes now require you to call a method. In particular:

• the port attribute of the GalaxyIO.Server class is now the ListenPort() method
• the validate attribute of the GalaxyIO.Server class is now the Validating() and EnableValidation() methods
• the server_name attribute of the GalaxyIO.Server class is now the ServerName() method
• the max_connections attribute of the GalaxyIO.Server class is now the MaxConnections() method

Version 2.x

s.max_connections = 5

s.MaxConnections(5)

Also, because Python now uses the C library:

• The method ReadFrameFromString() of the Galaxy.Frame class has been removed.
• The Galaxy.OPr() function no longer accepts the indent keyword.

Version 2.x

import os, sys

sys.path.insert(0, os.environ["LIBDIR"])

import Galaxy, GalaxyIO

import os, sys

sys.path.insert(0, os.path.join(os.environ["GC_HOME"],
                                "contrib", "MITRE", "templates"))

import GC_py_init

import Galaxy, GalaxyIO

Finally, the Python bindings now fully embrace the new call environments (see the documentation on adding a server and session management). The GalaxyIO.Connection class, which was "secretly" a call environment in 2.x, has now been renamed GalaxyIO.CallEnvironment.The impact of this change is minimal in non-class-based servers. We use a simplified version of the multiply server from the double example to illustrate:

Version 2.x

def Multiply(conn, dict):
    return {":int": dict[":int"] * Factor}

def Welcome(conn, dict):
    global Factor
    try:
        Factor = dict[":factor"]
    except: pass
    return None

def main():
    s = GalaxyIO.Server(sys.argv, "multiply", default_port = 2900,
           conn_class = GalaxyIO.Connection)
    s.AddDispatchFunction("multiply", Multiply)
    s.AddDispatchFunction("reinitialize", Welcome)
    s.RunServer()

main()

def Multiply(env, dict):
    return {":int": dict[":int"] * Factor}

def Welcome(env, dict):
    global Factor
    try:
        Factor = dict[":factor"]
    except: pass
    return None

def main():
    s = GalaxyIO.Server(sys.argv, "multiply", default_port = 2900)
    s.AddDispatchFunction("multiply", Multiply)
    s.AddDispatchFunction("reinitialize", Welcome)
    s.RunServer()

main()

Version 2.x

class MultiplyConnection(GalaxyIO.Connection):

    def Multiply(self, dict):
        return {":int": dict[":int"] * self.conn.Factor}

    def Welcome(self, dict):
        try:
            self.conn.Factor = dict[":factor"]
        except: self.conn.Factor = 1
        return None

def main():
    s = GalaxyIO.Server(sys.argv, "multiply", default_port = 2900,
                conn_class = MultiplyConnection)
    s.AddDispatchFunction("multiply", MultiplyConnection.Multiply)
    s.AddDispatchFunction("reinitialize", MultiplyConnection.Welcome)
    s.RunServer()

main()

class MultiplyEnvironment(GalaxyIO.CallEnvironment):

    def Multiply(self, dict):
        return {":int": dict[":int"] * self.conn.Factor}

    def Welcome(self, dict):
        try:
            self.conn.Factor = dict[":factor"]
        except: self.conn.Factor = 1
        return None

def main():
    s = GalaxyIO.Server(sys.argv, "multiply", default_port = 2900,
                env_class = MultiplyEnvironment)
    s.AddDispatchFunction("multiply", MultiplyEnvironment.Multiply)
    s.AddDispatchFunction("reinitialize", MultiplyEnvironment.Welcome)
    s.RunServer()

main()

Step 4: Upgrading Java bindings

Brokering

Java broker classes have been modified. First, the galaxy.server.DataOutBrokerconstructor has changed:

Version 2.x

     public DataOutBroker(Server server, int port, GFrame frame) throws Exception

     public DataOutBroker(Server server, int port, GFrame frame, boolean rport) throws Exception

Version 3.0

    public DataOutBroker(Server server, int pollMilliseconds, int timeoutSeconds) throws Exception

Java outgoing brokers now rely on the listener in the main server to handle broker client connection requests, so the broker no longer needs its own listener (i.e., port). Also, the GFrame is no longer needed (it provided the broker's call ID) since DataOutBroker now generates its own call ID (accessible via DataOutBroker.getCallId). The new constructor's arguments include a reference to the galaxy.server.Server that started the broker, a polling time in milliseconds, and a timeout in seconds. The polling time controls how often the broker checks for new data to send to its clients (it also checks for new clients, to which it sends any cached data it has). The timeout determines when (if ever) the broker stops accepting new client connections. Once a timeout expires, no new connections are accepted, and the broker will exit once it has been instructed to disconnect from its clients (via DataOutBroker.close). The old constructors are still available but have been deprecated. They now only use the Server parameter (default polling time and timeout values are used internally).

When sending messages from a server to a Hub, informing the Hub that brokered data is available, use  DataOutBroker.populateFrame to add the broker contact information automatically to the frame.

The codes segments below illustrate how code for creating an outgoing broker and creating a notification frame differs between Galaxy Communicator 2.x and 3.0.

Version 2.x

public class TestAudio extends galaxy.server.Server {
{
    ...

    Clause fr = new Clause("main");
    fr.setProperty(":call_id", this.getCallId());
    DataOutBroker broker=null;
    try {
         broker = new DataOutBroker(this, 3900, fr);
    } catch (Exception ex) {
         logAndSendError("Error opening broker connection: " + ex.toString());
         return;
    }

    ... // prepare brokered data

    GFrame result = new Clause("main");
    result.setProperty(":binary_port", broker.getPort());
    result.setProperty(":call_id", this.getCallId());
    result.setProperty(":binary_host", this.getHostAddress());

    ... // send the frame

Version 3.0

public class AudioOut extends galaxy.server.Server
{
    ...

    DataOutBroker broker;
    int pollMs = 50;
    int timeoutSecs = 50;
    try {
         broker = new DataOutBroker(this, pollMs, timeoutSecs);
    } catch (Exception e) {
         logAndSendError("Error opening broker connection: " + e.toString());
         return;
    }

    ... // prepare brokered data

    GFrame result = new Clause("main");
    broker.populateFrame(result, ":binary_host", ":binary_port");

    ... // send the frame

The manner in which an incoming broker detects and processes the end of a brokered data stream has also changed. Subclasses of galaxy.server.DataInBroker should implement the disconnectReceived method. This method is called automatically when the incoming broker receives a disconnect message from the outgoing broker, indicating that no more brokered data will be sent. Incoming brokers can also choose to implement abortReceived, which is called when the broker receives an unexpected exception while receiving brokered data.

Version 2.x

// Outgoing broker code segment

    DataOutBroker broker;

    ... // create broker

    broker.write("start");

    // write brokered data here

    broker.write("end");
    broker.close();
 
 

// Incoming broker code segment

public  void receivedString(String str)
{
     if (str.equals("start")) {
         System.out.println("About to start receiving data");
     } else if (str.equals("end")) {

     ... // process data

Version 3.0

// Outgoing broker code segment

    DataOutBroker broker;

    ... // create broker

    // write brokered data here

    broker.close();

Note that the outgoing broker no longer has to send explicit markers for the beginning and end of the brokered data.

// Incoming broker code segment

protected void disconnectReceived()
{
     ... // process data
 

As a result of these and other changes to brokering, there are a number of incompatabilities between 2.x and 3.0 code:
 

The galaxy.server.DataInBroker thread is no longer started in the DataInBroker constructor. Instead, DataInBroker.start must be called explicitly (e.g., at the end of a DataInBroker subclass constructor). This is required in order to ensure that a DataInBroker does not attempt to retrieve and process any brokered data until all custom initialization has taken place:

    public class AudioInBroker extends galaxy.server.DataInBroker
    {
        private GBinary bdata;

        public AudioInBroker(Server server, InetAddress ip, int port, GFrame frame)
            throws Exception
        {
             super(server,ip,port,frame);
             bdata = new GBinary();

             // You must explicitly call start() once your broker is initialized.
             // For example, you can call it at the end of the constructor, as is
             // done here.
             start();
        }
    .
    .
    .
 

In order to enable all Galaxy data types to be available via brokering, a DataInBroker must implement four additional methods: receivedList, receivedInteger, receivedFloat, and receivedSymbol.

 

 
 
 
 
 
 
 
 
 
 
 
 
 
 
 

Since out brokers now rely on the main server listener to receive client connection requests, the following DataOutBroker methods have been removed:

setPort

getPort

startListening

init

flush

stopListening

In addition, galaxy.server.Server.getCallID has been removed. Use DataOutBroker.getCallId instead.

Threads

Thread management has been redesigned in the Java bindings, especially with regards to stopping various threads. As a result, the following methods have been removed:

• galaxy.server.DataInBroker.close (use DataInBroker.stop to stop the broker)
• galaxy.server.MainServer.prepareToStop (use MainServer.stop to stop the main server thread)
• galaxy.server.Server.start (this did not do anything in 2.1, and it is not needed in 3.0)

• startListener (by default the listener thread, now encapsulated in galaxy.server.ListenerThread, is managed internally, but galaxy.server.MainServer.startListenerThread can be used to start the listener if it is not already running)
• connectToHubs (should not impact Galaxy Communicator developers)
• createServer (should not impact Galaxy Communicator developers)
• startServer (should not impact Galaxy Communicator developers)

Command Line Arguments

The syntax of command line arguments has changed in the Java bindings. Equal signs (=) are no longer needed when assigning values (e.g., -port=123 is now -port 123). Also, double quotes (") are now needed when listing multiple values for one argument (e.g., -contact_hub=localhost:123,localhost:456 is now -contact_hub "localhost:123 localhost:456").

The -noui argument has been replaced with -ui (i.e., graphical user interfaces based on ServerUI, if present, are now off by default; use -ui to turn them on).

Finally, the arguments appendLog and -mainServerClass have been renamed to -append_log and -main_server_class respectively.

See the Java binding documentation for details on new command line arguments.

Miscellaneous

This section contains information on other potential 2.x/3.0 incompatibilities in addition to those discussed above.

These classes have been removed from 3.0:

galaxy.server.WrongTypeInDispatch (should not impact Galaxy Communicator developers)

galaxy.io.GalaxyStreamable (should not impact Galaxy Communicator developers)

galaxy.io.GalaxyStream (should not impact Galaxy Communicator developers)

galaxy.util.HostPort (should not impact Galaxy Communicator developers)

galaxy.lang.Symbols (symbol definitions are now public static members of galaxy.lang.Symbol)

galaxy.lang.NullFrameException (should not impact Galaxy Communicator developers)

galaxy.server.MainServer.getAllowMultipleConnections

galaxy.server.MainServer.setAllowMultipleConnections (use the new -maxconns command line argument or MainServer.getMaxConnections and setMaxConnections to configure the number of maximum connections)

galaxy.server.MainServer.setHostPortCollection (this functionality has been replaced by MainServer.setHubContactInfoCollection, and there should not be any impact Galaxy Communicator developers)

galaxy.lang.Symbol.setOrdinal

galaxy.lang.Symbol.getOrdinal

galaxy.util.ArgParser.getArgCollections (should not impact Galaxy Communicator developers)

galaxy.io.GalaxyInputStream.write(ObjectStorage)(should not impact Galaxy Communicator developers)

galaxy.io.GalaxyInputStream.readBinary now returns galaxy.lang.GBinary instead of galaxy.lang.BinaryObject (which has been deprecated as of 3.0)

The galaxy.util.ArgParser.getArg methods now return java.util.List (was java.lang.String) since one command line argument can have a list of associated values (e.g., -contact_hub). In a related change, the version of getArg that takes a default argument value now expects a java.util.List (was java.lang.String).

The galaxy.lang.Gframeconstructor has been changed. It takes a galaxy.lang.Symbol and an integer that represents the frame type (types are now defined in galaxy.lang.GFrame: see  GAL_NULLFRAME, GAL_TOPIC, GAL_CLAUSE, GAL_PRED).

Finally, the Java method galaxy.server.Server.serverOpReinitialize is now public since the reinitialize method is treated like any other dispatch function in 3.0.

See the Galaxy Communicator release notes for additional information on changes and deprecations in the Java bindings.

Step 5: Upgrading server listen status

Version 2.1

if ((GalIO_ServerListenStatus(scomm) & GAL_SERVER_TYPE_MASK) == GAL_HUB_CLIENT))
  ...

if ((GalIO_ServerListenStatus(scomm) & GAL_SERVER_TYPE_MASK) == GAL_LISTENING_SERVER)
  ...

if (GalIO_ServerIsClient(scomm))
  ...

if (GalIO_ServerIsListener(scomm))
  ...

Step 6: Upgrading references to the call environment

In version 2.1, the call environment could "force a reply", which would cause a message to be sent to the Hub which counted as the reply to the Hub message which caused the dispatch function to fire. All subsequent messages which were written through the call environment, including the return value from the dispatch function, counted as new messages to the Hub. This functionality supported some idiosyncratic properties of the reinitialize dispatch function. We strongly discouraged people from exploiting this functionality in other cases. So for instance, while we also used this reply forcing technique to provide error and destroy replies to Hub messages, we advised programmers not to send any further "replies" through that call environment, since they would be treated as new messages. So the recommended use of GalSS_EnvError, for instance, was as follows:

Version 2.1

Gal_Frame dispatch_fn(Gal_Frame f, void *server_data)
{
  GalSS_Environment *env = (GalSS_Environment *) server _data;

  ...

  if (something_bad_happened) {
    GalSS_EnvError(env, "Something bad happened");
    return (Gal_Frame) NULL;
  }

  ...
}

Version 2.1

Gal_Frame dispatch_fn(Gal_Frame f, void *server_data)
{
  GalSS_Environment *env = (GalSS_Environment *) server _data;

  ...

  if (something_bad_happened) {
    GalSS_EnvError(env, "Something bad happened");
    GalSS_EnvError(env, "It was really bad");
    GalSS_SetProp(f, ":details", Gal_StringObject("Really, really bad"));
    return f;
  }

  ...
}

This behavior was incoherent and difficult to understand, and it has been changed in 3.0, as part of a general clarification and extension of call environments. In 3.0, a call environment will never send more than one return message to the Hub. Only four things count as return messages: a return value from a dispatch function, a call to GalSS_EnvError, a call to GalSS_EnvDestroyToken, or a call to GalSS_EnvReply. After the first reply is sent, all subsequent replies are ignored. So in the second example here, the second call to GalSS_EnvError will have no effect, and the frame returned from the dispatch function will be ignored.

Step 7: Upgrading calls to GalIO_SetServerData

As a consequence of this observation, we recommend that users now avoid returning data from _GalSS_init_server, as shown in this pre-3.0 example:

void *_GalSS_init_server(GalIO_ServerStruct *s, int argc, char **argv)
{
  int i, increment = 1;

  if (!GalUtil_OACheckUsage(argc, argv, oas, &i))
    exit(1);
  GalUtil_OAExtract(argc, argv, oas, "-increment", GAL_OA_INT, &increment);
  signal(SIGINT, exit_gracefully);
  /* Return value is server data. */
  return (void *) increment;
}

void *_GalSS_init_server(GalIO_ServerStruct *s, int argc, char **argv)
{
  int i, increment = 1;

  if (!GalUtil_OACheckUsage(argc, argv, oas, &i))
    exit(1);
  GalUtil_OAExtract(argc, argv, oas, "-increment", GAL_OA_INT, &increment);
  signal(SIGINT, exit_gracefully);
  GalIO_SetServerData(s, (void *) increment, NULL);
  return (void *) NULL;
}

Step 8a: Upgrading references to the Communicator main loop (optional)

int main(int argc, char **argv)
{
  GalIO_ServerStruct *server;

  server = GalSS_CmdlineInitializeServer(argc, argv);
  if (!server) {
    GalUtil_Fatal("Failed to initialize server!\n");
  }
  GalSS_RunServer(server);
  exit(0);
}

int main(int argc, char **argv)
{
  GalIO_ServerStruct *server;

  server = GalSS_CmdlineSetupServer(argc, argv);
  if (!server) {
    GalUtil_Fatal("Failed to set up server!\n");
  }
  GalSS_StartAndRunServer(server);
  exit(0);
}

Step 8b: Upgrade your own main loop (advanced, optional)

The main loop

• The main loop now sets up event callbacks which will set up the local callbacks, instead of doing it directly. This is because the server startup can happen in a number of different places.
• The server now has a connection event callback which handles the loop-specific initializations when a server accepts a connection.
• The new strategy reflects the fact that servers can now simultaneously be listeners and Hub clients.
• The server now sets up the server and later runs it explicitly, rather than initializing it (setting it up and running it), as discussed in this upgrade step. The new strategy gives us uniform access to the event callbacks.

Version 2.1

int main(int argc, char **argv)
{
  GalaxyCallbackRecord *gcr = (GalaxyCallbackRecord *) NULL;
  int new_argc;
  char **new_argv;
  GalSS_ServerArgs *arg_pkg;

  gcr = (GalaxyCallbackRecord *) malloc(sizeof(GalaxyCallbackRecord));
  gcr->timer_cb = (TimerCallback *) NULL;
  gcr->l = SM_NewLooper();

  /* If you want to use the built-in server arguments, you
     can use GalSS_ExtractServerArgs. Otherwise, you can just
     call GalSS_InitializeServerToplevel(). */

  arg_pkg = GalSS_DefaultServerArgs();

  /* Make sure it knows that we're using our own main loop. */
  GalSS_SAFixLoopType(arg_pkg, GAL_LOOP_EXTERNAL);
  arg_pkg = GalSS_ExtractServerArgs(arg_pkg, argc, argv,
                                    &new_argc, &new_argv);

  if (!arg_pkg) {
    /* Something bad happened, or -help was passed. */
    exit(1);
  }

  gcr->gcomm = GalSS_InitializeServerFromServerArgs(arg_pkg,
                                                    new_argc,
                                                    new_argv);
  GalSS_FreeArgPkg(arg_pkg);

  if (!gcr->gcomm) {
    fprintf(stderr, "Couldn't create a server\n");
    fflush(stderr);
    exit(1);
  }

  /* If the server is acting as a client, we'll want to
     set up callbacks for the connections; if it has its own
     listener, we'll want to set up a callback for the server,
     which will set up the listeners for the connections as
     they're received. */

  if ((GalIO_ServerListenStatus(gcr->gcomm) & GAL_SERVER_TYPE_MASK) ==
      GAL_LISTENING_SERVER) {
    gcr->server_sock = GalIO_GetServerListenSocket(gcr->gcomm);
    SM_AddFDCallback(gcr->l, gcr->server_sock, DoubleServerHandler,
                     (void *) gcr);
  } else {
    GalIO_ServerCheckHubContacts(gcr->gcomm);
    if (GalIO_GetServerNumConnections(gcr->gcomm) == 0) {
      fprintf(stderr, "Couldn't contact any Hubs\n");
      fflush(stderr);
      exit(1);
    }
    /* Add a callback for the connections. */
    GalIO_OperateOnConnections(gcr->gcomm, (void *) gcr,
                               __AddConnectionCallback);
  }
  SM_Mainloop(gcr->l);
  exit(0);
}

int main(int argc, char **argv)
{
  GalaxyCallbackRecord *gcr = (GalaxyCallbackRecord *) NULL;
  int new_argc;
  char **new_argv;
  GalSS_ServerArgs *arg_pkg;

  gcr = (GalaxyCallbackRecord *) malloc(sizeof(GalaxyCallbackRecord));
  gcr->timer_cb = (TimerCallback *) NULL;
  gcr->l = SM_NewLooper();

  /* If you want to use the built-in server arguments, you
     can use GalSS_ExtractCmdlineServerArgs. Otherwise, you can just
     call GalSS_InitializeServerToplevel(). */

  arg_pkg = GalSS_DefaultServerArgs();

  /* Make sure it knows that we're using our own main loop. We set this
   before we ever parse the server arguments, because we don't even want
   the arguments pertaining to the loop type enabled for the user. */

  GalSS_SAFixLoopType(arg_pkg, GAL_LOOP_EXTERNAL);
  arg_pkg = GalSS_ExtractCmdlineServerArgs(arg_pkg, argc, argv,
                                           &new_argc, &new_argv);

  if (!arg_pkg) {
    /* Something bad happened, or -help was passed. */
    exit(1);
  }

  /* Now, we call GalSS_InitializeServerFromServerArgs, and we don't have
     to worry about the signature of GalSS_InitializeServerToplevel. */

  gcr->scomm = GalSS_SetupServer(arg_pkg, new_argc, new_argv);
  GalSS_FreeArgPkg(arg_pkg);

  if (!gcr->scomm) {
    fprintf(stderr, "Couldn't create a server\n");
    fflush(stderr);
    exit(1);
  }

  /* Set the connect callback for the server. This gets called
     whenever a new connection is established. */

  GalIO_AddServerConnectCallback(gcr->scomm,
                                 GCRAddConnectionCallback, (void *) gcr);
  /* The server can be a listener when it starts out, or
     it can become a listener when an outgoing broker starts up. So
     we set a callback to handle whenever this happens. */
  GalIO_AddServerCallback(gcr->scomm,
                          GAL_SERVER_LISTENER_STARTUP_EVENT,
                          GCRSetupServerListener, (void *) gcr);
  /* Similarly, if someone calls GalIO_ContactHub, it may lead to
     a new poller starting up. So we should deal with that
     as a callback too. */
  GalIO_AddServerCallback(gcr->scomm,
                          GAL_SERVER_CLIENT_POLL_STARTUP_EVENT,
                          GCRSetupServerClient, (void *) gcr);
  /* And now, something that will shut down the loop when
     the server is destroyed. */
  GalIO_AddServerCallback(gcr->scomm,
                          GAL_SERVER_DESTRUCTION_EVENT,
                          GCRDestroyServer, (void *) gcr);

  /* Now, start the server, and then the main loop. */
  if (!GalIO_ServerStart(gcr->scomm)) {
    fprintf(stderr, "Couldn't start the server\n");
    fflush(stderr);
    exit(1);
  }
  SM_Mainloop(gcr->l);
  exit(0);
}

The server callbacks

• The startups for both client and listener are handled in parallel ways, using the event callbacks to set up local main loop callbacks as well as subsequent event callbacks (such as destruction callbacks).
• There's a more general callback handler (GalIO_ServerCallbackHandler) which calls the connect callback for the server. Therefore, our server callback handler doesn't need to set up the connection.
• GalIO_ServerCallbackHandler also takes care of destroying the server in the appropriate circumstances.
• Because the main loop requires both timer and file descriptor callbacks, both need to be cleared when the server dies.
• There's now a distinction between the server shutting down and the server being destroyed, which is reflected in an addition possible value returned from GalIO_ServerCallbackHandler. This is also reflected in the different event callbacks that are set up.

/* This function is only used when the server is running its
   own listener. */

static void DoubleServerHandler(void *client_data)
{
  int status;
  GalaxyCallbackRecord *gcr = (GalaxyCallbackRecord *) client_data;
  GalIO_CommStruct *new_conn = (GalIO_CommStruct *) NULL;

  status = GalIO_ServerHandler(gcr->gcomm, &new_conn);

  switch (status) {
  case 1:
    /* We've got a connection. */
    __AddConnectionCallback(new_conn, (void *) gcr);
    break;
  case -1:
    /* An error has occurred. */
    GalUtil_Warn("The server has failed.\n");
    SM_RemoveAllFDCallbacks(gcr->l);
    GalIO_SetServerDone(gcr->gcomm);
    GalIO_DestroyServerStruct(gcr->gcomm);
    break;
  default:
    break;
  }
}

static void GCRServerListenerHandler(void *client_data)
{
  GalaxyCallbackRecord *gcr = (GalaxyCallbackRecord *) client_data;

  GalIO_ServerCallbackHandler(gcr->scomm, 0,
                              (GalIO_CommStruct **) NULL);
}

static void GCRShutdownServerListener(GalIO_ServerStruct *scomm,
                                      void *callback_data)
{
  GalaxyCallbackRecord *gcr = (GalaxyCallbackRecord *) callback_data;

  GalUtil_Warn("The server has failed.\n");
  SM_RemoveAllFDCallbacks(gcr->l);
  SM_RemoveAllTimerCallbacks(gcr->l);
}

static void GCRSetupServerListener(GalIO_ServerStruct *scomm,
                                   void *callback_data)
{
  GalaxyCallbackRecord *gcr = (GalaxyCallbackRecord *) callback_data;
  /* You only need a file descriptor callback here, since
     there will be no connection requests in any internal queue. */
  SM_AddFDCallback(gcr->l, GalIO_GetServerListenSocket(scomm),
                   GCRServerListenerHandler,
                   (void *) gcr);
  GalIO_AddServerCallback(scomm,
                          GAL_SERVER_LISTENER_SHUTDOWN_EVENT,
                          GCRShutdownServerListener,
                          (void *) gcr);
}

/* This function is used when the server is subscribing to
   Hub listeners. */

static void GCRServerClientHandler(void *client_data)
{
  GalaxyCallbackRecord *gcr = (GalaxyCallbackRecord *) client_data;

  GalIO_ServerCheckHubContacts(gcr->scomm);
}

static void GCRShutdownServerClient(GalIO_ServerStruct *scomm,
                                    void *callback_data)
{
  TimerCallback *cb = (TimerCallback *) callback_data;

  SM_RemoveTimerCallback(cb);
}

static void GCRSetupServerClient(GalIO_ServerStruct *scomm,
                                 void *callback_data)
{
  GalaxyCallbackRecord *gcr = (GalaxyCallbackRecord *) callback_data;
  TimerCallback *cb;

  /* Set up a periodic task to check the hub contacts. */
  cb = SM_AddTimerCallback(gcr->l, 10, GCRServerClientHandler, (void *) gcr);
  /* Add a shutdown callback now. */
  GalIO_AddServerCallback(scomm,
                          GAL_SERVER_DESTRUCTION_EVENT,
                          GCRShutdownServerClient,
                          (void *) cb);
}

static void GCRDestroyServer(GalIO_ServerStruct *scomm,

                             void *callback_data)
{
  GalaxyCallbackRecord *gcr = (GalaxyCallbackRecord *) callback_data;

  GalUtil_Warn("The server has been destroyed.\n");
  SM_LooperExit(gcr->l);
}
 

The connection callbacks

• In 2.1, we called GalIO_ConnectionPoll, while in 3.0, we call the more general GalIO_ConnectionCallbackHandler.
• In 2.1, we shut down the server in the listener-in-Hub case when the last connection died, while we leave the client contact timer callback running in 3.0.
• In 2.1, we only used a file descriptor callback for connections, while we've clarified in 3.0 that a timer callback is also needed. This addresses the kludge reflected in the comment in 2.1 beginning "Sometimes, God help us, there's already data".
• In 3.0, we make use of the disconnect callback to shut down the file descriptor and timer callbacks for the connection, while in 2.1, we did it explicitly in the connection callback.
• In 3.0, we can also control broker startup with event callbacks, while that option was not available in 2.1.

static void DoubleConnectionHandler(void *client_data)
{
  GalIO_CommStruct *new_conn = (GalIO_CommStruct *) client_data;
  GalaxyCallbackRecord *gcr = (GalaxyCallbackRecord *) GalIO_GetCommData(new_conn);
  int fd = GalIO_GetCommSocket(new_conn);

  /* GalIO_ConnectionPoll returns 1 or -1 when the client
     should be terminated. */
  switch (GalIO_ConnectionPoll(new_conn)) {
  case 1:
    /* Done, stop polling. */
  case -1:
    /* Error, stop polling. */
    SM_RemoveFDCallback(gcr->l, fd);
    /* If this is the last connection, and the server
       is acting as a client, exit. */
    if (((GalIO_ServerListenStatus(gcr->gcomm) & GAL_SERVER_TYPE_MASK) ==
         GAL_HUB_CLIENT) &&
        ((GalIO_ServerListenStatus(gcr->gcomm) & GAL_HUB_CLIENT_DISCONNECT_MASK) ==
         GAL_HUB_CLIENT_DISCONNECT_SHUTDOWN) &&
        (SM_NumFDCallbacks(gcr->l) == 0)) {
      SM_LooperExit(gcr->l);
    }
    break;
  default:
    break;
  }
}

static void __AddConnectionCallback(GalIO_CommStruct *gcomm, void *arg)
{
  GalaxyCallbackRecord *gcr = (GalaxyCallbackRecord *) arg;
  int client_sock = GalIO_GetCommSocket(gcomm);

  GalIO_SetCommData(gcomm, gcr, NULL);
  SM_AddFDCallback(gcr->l, client_sock, DoubleConnectionHandler,
                   (void *) gcomm);

  /* Sometimes, God help us, there's already data coming in.
     Particularly if this is being called when the listener
     is in the Hub. */
  DoubleConnectionHandler((void *) gcomm);
}

typedef struct __connection_container {
  GalIO_CommStruct *gcomm;
  GalaxyCallbackRecord *gcr;
  TimerCallback *t;
  GAL_SOCKET fd;
} ConnectionContainer;

typedef struct __connection_container {
  GalIO_CommStruct *gcomm;
  GalaxyCallbackRecord *gcr;
  TimerCallback *t;
  GAL_SOCKET fd;
} ConnectionContainer;

/* GalIO_ConnectionCallbackHandler():
   -1 means an error was encountered and the connection has been destroyed.
   0 means we're in the midst of things.
   1 means we're done and the connection has been destroyed. */

static void GCRConnectionDisconnect(GalIO_CommStruct *gcomm, void *caller_data)
{
  ConnectionContainer *c = (ConnectionContainer *) caller_data;
  GalaxyCallbackRecord *gcr = c->gcr;

  SM_RemoveFDCallback(gcr->l, c->fd);
  SM_RemoveTimerCallback(c->t);
  free(c);
}

/* The loop cleanup is handled in the disconnect callback. */

static void GCRConnectionHandler(void *client_data)
{
  ConnectionContainer *c = (ConnectionContainer *) client_data;
  GalIO_CommStruct *gcomm = c->gcomm;

  GalIO_ConnectionCallbackHandler(gcomm, 0);
}

static void GCRConnectionTimerHandler(void *client_data)
{
  /* This is called from the timer. We could go ahead and
     just call the normal connection handler, which would
     try to read from the file descriptor, but since there's
     already a file descriptor callback which triggers that
     handler, we'll only do something if there's stuff
     in the internal queues. */
  ConnectionContainer *c = (ConnectionContainer *) client_data;
  GalIO_CommStruct *gcomm = c->gcomm;

  if (GalIO_CommReadReady(gcomm) || GalIO_CommWriteReady(gcomm)) {
    GCRConnectionHandler(client_data);
  }
}

static void GCRAddConnectionCallback(GalIO_ServerStruct *scomm,
                                     GalIO_CommStruct *gcomm,
                                     void *callback_data)
{
  GalaxyCallbackRecord *gcr = (GalaxyCallbackRecord *) callback_data;
  ConnectionContainer *c = (ConnectionContainer *) calloc(1, sizeof(ConnectionContainer));

  c->gcr = gcr;
  c->gcomm = gcomm;
  c->fd = GalIO_GetCommSocket(gcomm);

  /* We'll use the file descriptor callback to check the file
     descriptor, and the timer callback to check the internal queue. */
  SM_AddFDCallback(gcr->l, c->fd,
                   GCRConnectionHandler,
                   (void *) c);
  c->t = SM_AddTimerCallback(gcr->l, 5, GCRConnectionTimerHandler,
                             (void *) c);
  /* Finally, to support brokers, and to deal with
     disconnections, we need to use the
     data slot for the connection. */
  /* Make sure you stop polling when the connection dies. */
  GalIO_AddConnectionCallback(gcomm,
                              GAL_CONNECTION_SHUTDOWN_EVENT,
                              GCRConnectionDisconnect,
                              (void *) c);
  /* And now, add the callbacks for the broker setups. */
  GalIO_AddConnectionBrokerCallback(gcomm,
                                    GAL_CONNECTION_BROKER_OUT_STARTUP_EVENT,
                                    GCRSetupBrokerOut,
                                    (void *) gcr);
  GalIO_AddConnectionBrokerCallback(gcomm,
                                    GAL_CONNECTION_BROKER_IN_STARTUP_EVENT,
                                    GCRSetupBrokerIn,
                                    (void *) gcr);
}

Step 9: Upgrade timed task invocation (advanced, optional)

However, the timed task mechanism had a glaring shortcoming in version 2.1: while it was possible to associate both read and write sockets or both read or write file pointers with a timed task, it was not possible to tell which one led the task to be fired. Using the simpler tasks as an underlying representation, it was impossible to address this shortcoming; however, we were able to address it by "inverting" the implementation, making the complex tasks basic and implementing the simple tasks on top of them. Users of Gal_AddTask, etc., can now determine why a task was fired using the function Gal_TaskPkgRunReasons.

In version 3.0, we recommend that everyone upgrade to complex tasks. See the timed tasks documentation for full details on the functions. Here's the sort of change we recommend making (from the  MITRE stdin polling utility):

Version 2.1

static void stdin_poll(void *server_data);

Gal_AddTimedTaskWithFileIO(stdin_poll,
                           (void *) poll_struct,
                           poll_struct->ms,
                           stdin, (FILE *) NULL);

static void stdin_poll(Gal_TaskPkg *p);

Gal_AddTaskWithFileIO(stdin_poll,
                      (void *) poll_struct,
                      poll_struct->ms, 1,
                      stdin, (FILE *) NULL, NULL);

Version 2.1

static void stdin_poll(void *server_data)
{
  ...
  Gal_AddTimedTaskWithFileIO(stdin_poll,
                             (void *) poll_struct,
                             poll_struct->ms,
                             stdin, (FILE *) NULL);
  ...
}

static void stdin_poll(Gal_TaskPkg *p)
{
  ...
  Gal_ReAddTaskWithFileIO(p,
                          (void *) poll_struct,
                          poll_struct->ms, 1,
                          stdin, (FILE *) NULL, NULL);
  ...
}

Step 10: Upgrading session management (advanced, optional)

• While call environments embodied the dispatch function invocation context (including the session information), they couldn't be saved outside the context of their dispatch function (such as in broker callbacks or timed tasks).
• In order to forward the session information appropriately, the programmer needed to store away the session ID and add it to any new messages which were sent from broker callbacks or timed tasks.
• The behavior of call environments was idiosyncratic in other ways (see, for instance, the section on upgrading call environments for version 3.0).

Without environments

static void audio_handler(GalIO_BrokerStruct *broker_struct,
                          void *data, Gal_ObjectType data_type,
                          int n_samples)
{
  DataHandler *d = (DataHandler *) GalIO_GetBrokerCallerData(broker_struct);
  Gal_Frame f;

  ...

      f = Gal_MakeFrame("notify", GAL_CLAUSE);
      Gal_SetProp(f, ":notification", Gal_StringObject("Audio received."));
      GalIO_CommWriteFrame(d->gcomm, f, 0);
      Gal_FreeFrame(f);

  ...
}

void __FreeDataHandler(GalIO_BrokerStruct *b, void *data);

Gal_Frame receive_audio(Gal_Frame f, void *server_data)
{
  DataHandler *d = (DataHandler *) malloc(sizeof(DataHandler));
  GalIO_BrokerStruct *b;
  char *host = Gal_GetString(f, ":binary_host");
  int port = Gal_GetInt(f, ":binary_port");

  d->data_buf = (char *) NULL;
  d->size = 0;
  d->gcomm = GalSS_EnvComm((GalSS_Environment *) server_data);

  if (host && port) {
     b = GalIO_CommBrokerDataInInit(d->gcomm, host, port, f, audio_handler,
                                    0, d, __FreeDataHandler);
    if (b) {
      GalIO_SetBrokerActive(b);
    }
  } else {
    free(d);
  }
  return (Gal_Frame) NULL;
}

static void env_audio_handler(GalIO_BrokerStruct *broker_struct,
                              void *data, Gal_ObjectType data_type,
                              int n_samples)
{
  DataHandler *d = (DataHandler *) GalIO_GetBrokerCallerData(broker_struct);
  Gal_Frame f;
  GalSS_Environment *env = GalSS_BrokerGetEnvironment(broker_struct);

  ...

      f = Gal_MakeFrame("notify", GAL_CLAUSE);
      Gal_SetProp(f, ":notification", Gal_StringObject("Audio received."));
      GalSS_EnvWriteFrame(env, f, 0);
      Gal_FreeFrame(f);

  ...
}

void __FreeDataHandler(GalIO_BrokerStruct *b, void *data);

Gal_Frame receive_audio(Gal_Frame f, void *server_data)
{
  DataHandler *d = (DataHandler *) malloc(sizeof(DataHandler));
  GalIO_BrokerStruct *b;
  char *host = Gal_GetString(f, ":binary_host");
  int port = Gal_GetInt(f, ":binary_port");

  d->data_buf = (char *) NULL;
  d->size = 0;

  if (host && port) {
    b = GalSS_EnvBrokerDataInInit((GalSS_Environment *) server_data,
                                  host, port, f,
                                  env_audio_handler, 0, d, __FreeDataHandler);
    if (b) {
      GalIO_SetBrokerActive(b);
    }
  } else {
    free(d);
  }
  return (Gal_Frame) NULL;
}

• Gal_AddTask can be replaced with GalSS_EnvAddTask, for those cases where environments are needed for writing in timed tasks.
• The stdin polling function MGalIO_CreateStdinPoll should be replaced with MGalSS_EnvCreateStdinPoll in all cases where the stdin callback returns a frame, because that frame will be written as a new message to the Hub.

License / Documentation home / Help and feedback

Last updated January 17, 2001