L:/channel/channel/examples/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 string IdType;

typedef Channel<IdType> Chan;
typedef Chan::Msg ChanMsg;

//------------------------------------------------------------------
// Message definitions :
//
//ids for channel using int as msg id/type 
/*
int PING_MSG=2000;
int PONG_MSG=2001;
int TEST_STRING_MSG=2002;
*/
//ids for channel using string as msg id/type 
std::string PING_MSG = "_PING_";
std::string PONG_MSG = "_PONG_";
std::string TEST_STRING_MSG = "_TEST_";

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 << "text_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: recv my_unix_sock_path peer_unix_sock_path\n";
    return -1;
  }

  //step1. create channel
  Chan * my_chan = new Chan(); 
  
  //step2. setup channels
  char *my_addr = argv[1];
  Chan::UnixSockConnector *unix_conn = new Chan::UnixSockConnector(my_chan, true);
  if (unix_conn->open (my_addr) == FAILURE) {
    cout << "failed to open my unix domain socket at " <<  my_addr << endl;
    return -1;
  }
  if (argc >= 3) {
    char *peer_addr = argv[2];
    if (unix_conn->connect (peer_addr) == FAILURE) { 
      cout << "failed to connect peer unix domain socket 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(TEST_STRING_MSG);
  my_calbak.subscribe_msg(TEST_STRING_MSG);

  cout << "text_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) {
      //
      if (msg->type == TEST_STRING_MSG) {
        sm = (Test_String_Msg *)msg->data();
        sm->data[sm->len-1] = '\0';

        cout << "text_recv receive the following: " <<  sm->data << endl;
        delete msg;
      }
    } else {
      //handle_error();
      break;
    }
  }

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

  return 0;
}

---