src/ipc/Port.cc

   1 /*
   2  * $Id$
   3  *
   4  * DEBUG: section 54    Interprocess Communication
   5  *
   6  */
   7
   8
   9 #include "config.h"
  10 #include "comm.h"
  11 #include "CommCalls.h"
  12 #include "comm/Connection.h"
  13 #include "ipc/Port.h"
  14
  15 const char Ipc::coordinatorAddr[] = DEFAULT_STATEDIR "/coordinator.ipc";
  16 const char Ipc::strandAddrPfx[] = DEFAULT_STATEDIR "/kid";
  17
  18
  19 Ipc::Port::Port(const String& aListenAddr):
  20         UdsOp(aListenAddr)
  21 {
  22     setOptions(COMM_NONBLOCKING | COMM_DOBIND);
  23 }
  24
  25 void Ipc::Port::start()
  26 {
  27     UdsOp::start();
  28     listen();
  29 }
  30
  31 void Ipc::Port::listen()
  32 {
  33     debugs(54, 6, HERE);
  34     buf.prepForReading();
  35     typedef CommCbMemFunT<Port, CommIoCbParams> Dialer;
  36     AsyncCall::Pointer readHandler = JobCallback(54, 6,
  37                                      Dialer, this, Port::noteRead);
  38     comm_read(conn(), buf.raw(), buf.size(), readHandler);
  39 }
  40
  41 bool Ipc::Port::doneAll() const
  42 {
  43     return false; // listen forever
  44 }
  45
  46 String Ipc::Port::MakeAddr(const char* pathAddr, int id)
  47 {
  48     assert(id >= 0);
  49     String addr = pathAddr;
  50     addr.append('-');
  51     addr.append(xitoa(id));
  52     addr.append(".ipc");
  53     return addr;
  54 }
  55
  56 void Ipc::Port::noteRead(const CommIoCbParams& params)
  57 {
  58     debugs(54, 6, HERE << params.conn << " flag " << params.flag <<
  59            " [" << this << ']');
  60     if (params.flag == COMM_OK) {
  61         assert(params.buf == buf.raw());
  62         receive(buf);
  63     }
  64     // TODO: if there was a fatal error on our socket, close the socket before
  65     // trying to listen again and print a level-1 error message.
  66
  67     listen();
  68 }