L:/channel/channel/examples/hier_id_text_recv/recv.cpp

Go to the documentation of this file.

// Copyright (c) 2005, 2006 Yigong Liu
// Permission to use, copy, modify, distribute and sell this software for any 
//     purpose is hereby granted without fee, provided that the above copyright 
//     notice appear in all copies and that both that copyright notice and this 
//     permission notice appear in supporting documentation.
// The author makes no representations about the 
//     suitability of this software for any purpose. It is provided "as is" 
//     without express or implied warranty.

#include <iostream>
#include <string>

#include <Channel.h>

using namespace std;
using namespace channel;

//---------------------------------------------------------------
// Channel specialization
//
typedef StringPathId<'/'> IdType;

typedef Channel<IdType, IdTrait<IdType>, ACE_MT_SYNCH, TrieRouter<IdType, IdTrait<IdType>, ACE_MT_SYNCH> > Chan;
typedef Chan::Msg ChanMsg;

//------------------------------------------------------------------
// Message definitions :
//
//app msgs ids 
IdType PING_MSG = "/APP/PING";
IdType PONG_MSG = "/APP/PONG";
IdType TEST_STRING_MSG = "/APP/TEST";
IdType APP_WILDCARD_MSG = "/APP/*";

struct Test_String_Msg {
  enum { MAX_STR_LEN = 1024 };
  int len;
  char data[MAX_STR_LEN];
  Test_String_Msg() {
  }
};

//----------------------------------------------------------------
// Implementation
//

class My_Callback: public Chan::Callback {
public:
  My_Callback(Chan *c) : Chan::Callback(c) {}
  Status process(Msg *msg) {
    Test_String_Msg *sm = (Test_String_Msg *)msg->data();
    cout << "hier_recv/My_Callback recv the following: " << sm->data << endl;
    cout << "...... Please notice callback borrow Channel internal thread\n";
    delete msg;
    return SUCCESS;
  }
};

int main (int argc, char **argv) {
  if (argc < 2) {
    cout << "Usage: hrecv my_host:port [peer_host:port]\n";
    return -1;
  }

  //step1. create channel
  Chan * my_chan = new Chan(); 
  
  //step2. setup channels
  char *my_addr = argv[1];
  Chan::TcpSockConnector *tcp_conn = new Chan::TcpSockConnector(my_chan, true);
  if (tcp_conn->open (my_addr) == FAILURE) {
    cout << "failed to open at " << my_addr << endl;
    return -1;
  }
  if (argc >= 3) {
    char *peer_addr = argv[2];
    ConnInfo ci(peer_addr);
    if (tcp_conn->connect (ci) == FAILURE) { 
      cout << "failed to connect peer  at " << peer_addr << endl;
      return -1;
    }
  }

  //step3. create ports/callbacks attached to channels
  Chan::Port my_port(my_chan); //use my own queue
  My_Callback my_calbak(my_chan);
 
  //step4. sub msgs thru ports/callbacks
  my_port.subscribe_msg(APP_WILDCARD_MSG);
  my_calbak.subscribe_msg(TEST_STRING_MSG);

  cout << "hier_recv coming up ...\n";

  ChanMsg *msg;
  Test_String_Msg *sm;
 
  //step5. recv msgs from ports and process them
  for(;;) {
    if(my_port.recv_msg(msg) == SUCCESS) {
      //
      sm = (Test_String_Msg *)msg->data();
      sm->data[sm->len-1] = '\0';
      cout << "hier_recv thread receive the following: " << sm->data << endl;
      delete msg;
    } else {
      //handle_error();
      break;
    }
  }

  cout << "hier_recv  exits...\n";

  return 0;
}

---