Android SocketListener、FrameworkListener 分析

logd.rc 中 有 "socket logd stream 0666 logd logd" 以及 "socket logdr seqpacket 0666 logd logd"

这样的写法，实际会将logd 服务 与  "/dev/socket/logd" "dev/socket/logdr" socket节点进行绑定

​/system/core/logd/logd.rc
service logd /system/bin/logd
    socket logd stream 0666 logd logd
    socket logdr seqpacket 0666 logd logd
    socket logdw dgram+passcred 0222 logd logd
    file /proc/kmsg r
    file /dev/kmsg w
    user logd
    group logd system package_info readproc
    capabilities SYSLOG AUDIT_ConTROL
    priority 10
    task_profiles ServiceCapacityLow

1) 通过继承 SocketListener 类，

SocketListener, class LogReader : public SocketListener {}"

2) 实现构造方法:

LogListener::LogListener(LogBuffer* buf, LogReader* reader)
    : SocketListener(getLogSocket(), false), logbuf(buf), reader(reader) {}

3) 通过 startListener()  以及重载的 onDataAvailable()方法 完成通信

system/core/libsysutils/include/sysutils/SocketListener.h
class SocketListener {
    bool                    mListen;
    const char              *mSocketName;
    int                     mSock;
    std::unordered_map mClients;
    pthread_mutex_t         mClientsLock;
    int                     mCtrlPipe[2];
    pthread_t               mThread;
    bool                    mUseCmdNum;

public:
    SocketListener(const char *socketName, bool listen);
    SocketListener(const char *socketName, bool listen, bool useCmdNum);
    SocketListener(int socketFd, bool listen);

    virtual ~SocketListener();
    int startListener();
    int startListener(int backlog);
    int stopListener();

    void sendBroadcast(int code, const char *msg, bool addErrno);

    void runonEachSocket(SocketClientCommand *command);

    bool release(SocketClient *c) { return release(c, true); }

protected:
    virtual bool onDataAvailable(SocketClient *c) = 0;

private:
    static void *threadStart(void *obj);

    // Add all clients to a separate list, so we don't have to hold the lock
    // while processing it.
    std::vector snapshotClients();

    bool release(SocketClient *c, bool wakeup);
    void runListener();
    void init(const char *socketName, int socketFd, bool listen, bool useCmdNum);
};

natvie 代码里面通过 android_get_control_socket(xxx) 获取 "/dev/socket/xxx"

/system/core/libsysutils/src/SocketListener.cpp

SocketListener::SocketListener(const char *socketName, bool listen) {
    init(socketName, -1, listen, false)
}
SocketListener::SocketListener(int socketFd, bool listen) {
  init(nullptr, socketFd, listen, false)
}

void SocketListener::init(const char *socketName, int socketFd, bool listen, bool 
  useCmdNum) {
  mListen = listen
  mSocketName = socketName
  mSock = socketFd
  mUseCmdNum = useCmdNum
  pthread_mutex_init(&mClientsLock, nullptr)
}

int SocketListener::startListener() {
  return startListener(4)
}

int SocketListener::startListener(int backlog) {
  if (mSocketName) {
    // 重要，获取 socket 的地方，通过 mSocketName 获取socket
    mSock = android_get_control_socket(mSocketName))
    fcntl(mSock, F_SETFD, FD_CLOEXEC)
  }

  if (mListen && listen(mSock, backlog) < 0)
  else if(!mListen)
    mClients[mSock] = new SocketClient(mSock, false, mUseCmdNum)
    
  (pthread_create(&mThread, nullptr, SocketListener::threadStart, this)
}
  
void *SocketListener::threadStart(void *obj) {
  SocketListener *me = reinterpret_cast(obj)
  me->runListener()
}

// 继承于 SocketListener 的子类，通过重载 onDataAvailable()方法，基于这个模型实现通信的
void SocketListener::runListener() {
  while (true) {
    for (SocketClient* c : pending) {
      onDataAvailable(c)
    }
  }
}

frameworkListener 继承于 SocketListener ，也可以基于 frameworkListener 进行通信设计

通过 registerCmd 方法注册，然后调用对应 frameworkCommand 的runCommand()方法

system/core/libsysutils/include/sysutils/frameworkListener.h
class frameworkListener : public SocketListener {}

/system/core/libsysutils/src/frameworkListener.cpp

frameworkListener::frameworkListener(const char *socketName, bool withSeq) :
  SocketListener(socketName, true, withSeq) {
  init(socketName, withSeq)
}

frameworkListener::frameworkListener(const char *socketName) :
  SocketListener(socketName, true, false) {
  init(socketName, false)
}

void frameworkListener::init(const char* , bool withSeq) {
  mWithSeq = withSeq
  mSkipTonextNullByte = false
}

bool frameworkListener::onDataAvailable(SocketClient *c) {
  len = TEMP_FAILURE_RETRY(read(c->getSocket(), buffer, sizeof(buffer)))
  for (i = 0; i < len; i++) {
    if (buffer[i] == '') {
       dispatchCommand(c, buffer + offset)
       offset = i + 1
    }
  }
}

void frameworkListener::dispatchCommand(SocketClient *cli, char *data) {
  while(*p) {
    for (auto* c : mCommands) {
      if (!strcmp(argv[0], c->getCommand())) {
        if (c->runCommand(cli, argc, argv))
      }
    }
  }
}

// 通过 registerCmd 方法注册，然后调用对应 frameworkCommand 的runCommand()方法
void frameworkListener::registerCmd(frameworkCommand *cmd) {
  mCommands.push_back(cmd)
}

// 继承 SocketListener 实现通信
/system/core/logd/LogListener.cpp
  LogListener::LogListener(LogBuffer* buf, LogReader* reader)
    : SocketListener(getLogSocket(), false), logbuf(buf), reader(reader) {}
  
  int LogListener::getLogSocket() {
    static const char socketName[] = "logdw";
    int sock = android_get_control_socket(socketName);
    if (sock < 0) {
      sock = socket_local_server(
        socketName, ANDROID_SOCKET_NAMESPACE_RESERVED, SOCK_DGRAM);
      setsockopt(sock, SOL_SOCKET, SO_PASSCRED, &on, sizeof(on))
    }
    return sock
  }

  bool LogListener::onDataAvailable(SocketClient* cli) {
  }

/system/core/logd/CommandListener.cpp
  int CommandListener::ClearCmd::runCommand(SocketClient* cli, int argc,char** argv) {
    if (argc < 2) {
      cli->sendMsg("Missing Argument");
      return 0;
    }
    cli->sendMsg(mBuf.clear((log_id_t)id, uid) ? "busy" : "success")
    return 0
  }

// 服务端，守护进程 logd

/system/core/logd/main.cpp
  int main(int argc, char* argv[]) {
    // 继承于 SocketListener, class LogReader : public SocketListener {}
    LogReader* reader = new LogReader(logBuf)
    reader->startListener()
    
    // 继承于 SocketListener, class LogListener : public SocketListener {}
    LogListener* swl = new LogListener(logBuf, reader)
    swl->startListener(600)
    
    // 继承于frameworkListener, class CommandListener : public frameworkListener {}
    CommandListener* cl = new CommandListener(logBuf, reader, swl)
    cl->startListener() 
  }

// 使用者客户端， logcat 

/system/core/logcat/logcat.cpp
int main(int argc, char** argv) {
  return logcat.Run(argc, argv)
}
int Logcat::Run(int argc, char** argv) {
  while (!max_count_ || print_count_ < max_count_) {
    int ret = android_logger_list_read(logger_list.get(), &log_msg)
  }
}

/system/core/liblog/logd_reader.cpp
int LogdRead(struct logger_list* logger_list, struct log_msg* log_msg) {
  int ret = logdOpen(logger_list)
}

static int logdOpen(struct logger_list* logger_list) {
  sock = socket_local_client("logdr", SOCK_SEQPACKET)
  ret = TEMP_FAILURE_RETRY(write(sock, buffer, cp - buffer))
}

static int socket_local_client(const std::string& name, int type) {
  std::string path = "/dev/socket/" + name;
  strlcpy(addr.sun_path, path.c_str(), sizeof(addr.sun_path))
  int fd = socket(AF_LOCAL, type | SOCK_CLOEXEC, 0)
  connect(fd, reinterpret_cast(&addr), sizeof(addr))
}