com::apama::engine::EngineManagement

/**
* Connect this Event Correlator as an event receiver to another
* Event Correlator.
*
* @param target The Correlator to connect to
* @param channels An array of names representing the channels to subscribe
* to. This is a null-terminated array of pointers to zero-terminated
* char arrays. If this is null or empty, subscribe to all channels.
* @return true if successful
* @exception EngineException
*/
virtual bool attachAsEventConsumerTo(
    EngineManagement* target, const char* const* channels)
    =0;
/**
* Unsubscribe as an event receiver from another Event Correlator.
*
* @param target The Correlator to unsubscribe from.
* @param channels An array of names representing the channels to
* unsubscribe from. This is a null-terminated array of pointers to
* zero-terminated char arrays. If this is null or empty unsubscribe
* from all channels.
* @exception EngineException
*/
virtual void detachAsEventConsumerFrom(
    EngineManagement* target, const char* const* channels)
     =0;

  // Method that connects to 2 ‘Engines’ (Correlators) and links them up
  // in a pipeline
  void connect(const char* host1, unsigned short port1,
    const char* host2, unsigned short port2,
    vector<const char*> channels, bool detach,
    int& rc )
  {
    const char* emsg = "";
    EngineManagement* engine1 = NULL;
    EngineManagement* engine2 = NULL;
    try {
      // a bit verbose
      cerr << "Requesting connection from host:" << host1
        << ", port:" << port1 << " to host:" << host2
        << ", port:" << port2 << endl;
      // Ensure valid arguments
      rc = 3; // this just sets a return value
      if ((host1==host2) && (port1==port2)) {
        cerr << "Connecting an engine to itself is disallowed"
          << endl;
        return;
    }
      // Make sure the channel list is NULL-terminated
      channels.push_back(NULL);
      // Attempt to connect to source Correlator
      rc = 1;
      emsg = "Failed to connect to source engine";
      if (!(engine1 =
        com::apama::engine::connectToEngine(host1, port1))) {
        throw EngineException(emsg);
      }
      // Attempt to connect to target Correlator
      rc = 2;
      emsg = "Failed to connect to target engine";
      if (!(engine2 =
        com::apama::engine::connectToEngine(host2, port2))) {
        throw EngineException(emsg);
      }
      // Connect target to source
      if (detach) {
        emsg = "Detach failed";
        engine2->detachAsEventConsumerFrom(engine1,&channels[0]);
      }
      else {
        emsg = "Attach failed";
        if (
          !engine2->attachAsEventConsumerTo(engine1,&channels[0]))
        {
          rc = 4;
          emsg = "Target engine could not connect to source engine";
          throw EngineException(emsg);
        }
      }
    }
    catch (EngineException ex) {
      string errmes;
      errmes += emsg;
      errmes += ": ";
      errmes += ex.what();
      throw EngineException(errmes.c_str());
    }
    // Shutdown cleanly
    rc = 5;
    if (engine1) {
      // Disconnect from Apama
      emsg = "Failed to disconnect from source engine";
      com::apama::engine::disconnectFromEngine(engine1);
    }
    if (engine2) {
      // Disconnect from Apama
      emsg = "Failed to disconnect from target engine";
      com::apama::engine::disconnectFromEngine(engine2);
    }
  }