Android中的socket本地通讯框架

一、先分析Native层：

1、C++基类SocketListener：

1. class SocketListener {
2.     int mSock;
3.     const
   char *mSocketName;
4.     SocketClientCollection *mClients;
5.     pthread_mutex_t mClientsLock;
6.     bool mListen;
7.     int mCtrlPipe[2];
8.     pthread_t mThread;
10. public:
11.     SocketListener(const
    char *socketNames, bool listen);
12.     SocketListener(int socketFd, bool listen);
14.     virtual ~SocketListener();
15.     int startListener();
16.     int stopListener();
18.     void sendBroadcast(int code, const
    char *msg, bool addErrno);
19.     void sendBroadcast(const
    char *msg);
21. protected:
22.     virtual bool onDataAvailable(SocketClient *c) = 0;
24. private:
25.     static
    void *threadStart(void *obj);
26.     void runListener();
27. };
28. #endif

看关键接口runListener：

1. void SocketListener::runListener() {
3.     while(1) {
4.         SocketClientCollection::iterator it;
5.         fd_set read_fds;
6.         int rc = 0;
7.         int max = 0;
9.         FD_ZERO(&read_fds);
11.         if (mListen) {
12.             max = mSock;
13.             FD_SET(mSock, &read_fds);
14.         }
16.         FD_SET(mCtrlPipe[0], &read_fds);
17.         if (mCtrlPipe[0] > max)
18.             max = mCtrlPipe[0];
20.         pthread_mutex_lock(&mClientsLock);
21.         for (it = mClients->begin(); it != mClients->end(); ++it) {
22.             FD_SET((*it)->getSocket(), &read_fds);
23.             if ((*it)->getSocket() > max)
24.                 max = (*it)->getSocket();
25.         }
26.         pthread_mutex_unlock(&mClientsLock);
28.         if ((rc = select(max + 1, &read_fds, NULL, NULL, NULL)) < 0) {
29.             SLOGE("select failed (%s)", strerror(errno));
30.             sleep(1);
31.             continue;
32.         } else
    if (!rc)
33.             continue;
35.         if (FD_ISSET(mCtrlPipe[0], &read_fds))
36.             break;
37.         if (mListen && FD_ISSET(mSock, &read_fds)) {
38.             struct sockaddr addr;
39.             socklen_t alen = sizeof(addr);
40.             int c;
42.             if ((c = accept(mSock, &addr, &alen)) < 0) {
43.                 SLOGE("accept failed (%s)", strerror(errno));
44.                 sleep(1);
45.                 continue;
46.             }
47.             pthread_mutex_lock(&mClientsLock);
48.             mClients->push_back(new SocketClient(c));
49.             pthread_mutex_unlock(&mClientsLock);
50.         }
52.         do {
53.             pthread_mutex_lock(&mClientsLock);
54.             for (it = mClients->begin(); it != mClients->end(); ++it) {
55.                 int fd = (*it)->getSocket();
56.                 if (FD_ISSET(fd, &read_fds)) {
57.                     pthread_mutex_unlock(&mClientsLock);
58.                     if (!onDataAvailable(*it)) {//由子类实现的接口。
59.                         close(fd);
60.                         pthread_mutex_lock(&mClientsLock);
61.                         delete *it;
62.                         it = mClients->erase(it);
63.                         pthread_mutex_unlock(&mClientsLock);
64.                     }
65.                     FD_CLR(fd, &read_fds);
66.                     continue;
67.                 }
68.             }
69.             pthread_mutex_unlock(&mClientsLock);
70.         } while (0);
71.     }
72. }

2、他的第一继承者：

1. #include "SocketListener.h"
2. #include "FrameworkCommand.h"
4. class SocketClient;
6. class FrameworkListener : public SocketListener {
7. public:
8.     static
   const
   int CMD_ARGS_MAX = 16;
9. private:
10.     FrameworkCommandCollection *mCommands;
12. public:
13.     FrameworkListener(const
    char *socketName);
14.     virtual ~FrameworkListener() {}
16. protected:
17.     void registerCmd(FrameworkCommand *cmd);
18.     virtual
    bool onDataAvailable(SocketClient *c);
20. private:
21.     void dispatchCommand(SocketClient *c, char *data);
22. };

他的onDataAvailable接口实现：

1. bool FrameworkListener::onDataAvailable(SocketClient *c) {
2.     char buffer[255];
3.     int len;
5.     if ((len = read(c->getSocket(), buffer, sizeof(buffer) -1)) < 0) {
6.         SLOGE("read() failed (%s)", strerror(errno));
7.         return errno;
8.     } else
   if (!len)
9.         return
   false;
11.     int offset = 0;
12.     int i;
14.     for (i = 0; i < len; i++) {
15.         if (buffer[i] == '\0') {
16.             dispatchCommand(c, buffer + offset);
17.             offset = i + 1;
18.         }
19.     }
20.     return
    true;
21. }

3、实例：netd的CommandListener类：

1. class CommandListener : public FrameworkListener {
2.     static TetherController *sTetherCtrl;
3.     static NatController *sNatCtrl;
4.     static PppController *sPppCtrl;
5.     static PanController *sPanCtrl;
6.     static SoftapController *sSoftapCtrl;
7.     static UsbController *sUsbCtrl;
9. public:
10.     CommandListener();
11.     virtual ~CommandListener() {}
13. private:
15.     static
    int readInterfaceCounters(const
    char *iface, unsigned long *rx, unsigned long *tx);
17.     class UsbCmd : public NetdCommand {
18.     public:
19.         UsbCmd();
20.         virtual ~UsbCmd() {}
21.         int runCommand(SocketClient *c, int argc, char ** argv);
22.     };
24.     class SoftapCmd : public NetdCommand {
25.     public:
26.         SoftapCmd();
27.         virtual ~SoftapCmd() {}
28.         int runCommand(SocketClient *c, int argc, char ** argv);
29.     };
31.     class InterfaceCmd : public NetdCommand {
32.     public:
33.         InterfaceCmd();
34.         virtual ~InterfaceCmd() {}
35.         int runCommand(SocketClient *c, int argc, char ** argv);
36.     };
38.     class IpFwdCmd : public NetdCommand {
39.     public:
40.         IpFwdCmd();
41.         virtual ~IpFwdCmd() {}
42.         int runCommand(SocketClient *c, int argc, char ** argv);
43.     };
45.     class TetherCmd : public NetdCommand {
46.     public:
47.         TetherCmd();
48.         virtual ~TetherCmd() {}
49.         int runCommand(SocketClient *c, int argc, char ** argv);
50.     };
52.     class NatCmd : public NetdCommand {
53.     public:
54.         NatCmd();
55.         virtual ~NatCmd() {}
56.         int runCommand(SocketClient *c, int argc, char ** argv);
57.     };
59.     class ListTtysCmd : public NetdCommand {
60.     public:
61.         ListTtysCmd();
62.         virtual ~ListTtysCmd() {}
63.         int runCommand(SocketClient *c, int argc, char ** argv);
64.     };
66.     class PppdCmd : public NetdCommand {
67.     public:
68.         PppdCmd();
69.         virtual ~PppdCmd() {}
70.         int runCommand(SocketClient *c, int argc, char ** argv);
71.     };
73.     class PanCmd : public NetdCommand {
74.     public:
75.         PanCmd();
76.         virtual ~PanCmd() {}
77.         int runCommand(SocketClient *c, int argc, char ** argv);
78.     };
79. };

不能忘记NetdCommand类：

1. #include <sysutils/FrameworkCommand.h>
3. class NetdCommand : public FrameworkCommand {
4. public:
5.     NetdCommand(const
   char *cmd);
6.     virtual ~NetdCommand() {}
7. };
8. 4、分析一个子类ListTtysCmd：
9. CommandListener::ListTtysCmd::ListTtysCmd() :
10.                  NetdCommand("list_ttys") {
11. }
13. int CommandListener::ListTtysCmd::runCommand(SocketClient *cli,
14.                                              int argc, char **argv) {
15.     TtyCollection *tlist = sPppCtrl->getTtyList();
16.     TtyCollection::iterator it;
18.     for (it = tlist->begin(); it != tlist->end(); ++it) {
19.         cli->sendMsg(ResponseCode::TtyListResult, *it, false);
20.     }
22.     cli->sendMsg(ResponseCode::CommandOkay, "Ttys listed.", false);
23.     return 0;
24. }

Java层：

1. private
   void listenToSocket() throws IOException {
2.     LocalSocket socket = null;
4.     try {
5.         socket = new LocalSocket();
6.         LocalSocketAddress address = new LocalSocketAddress(mSocket,
7.                 LocalSocketAddress.Namespace.RESERVED);
9.         socket.connect(address);
11.         InputStream inputStream = socket.getInputStream();
12.         mOutputStream = socket.getOutputStream();
14.         mCallbacks.onDaemonConnected();
16.         byte[] buffer = new
    byte[BUFFER_SIZE];
17.         int start = 0;
19.         while (true) {
20.             int count = inputStream.read(buffer, start, BUFFER_SIZE - start);
21.             if (count < 0) break;
23.             // Add our starting point to the count and reset the start.
24.             count += start;
25.             start = 0;
27.             for (int i = 0; i < count; i++) {
28.                 if (buffer[i] == 0) {
29.                     String event = new String(buffer, start, i - start);
30.                     if (LOCAL_LOGD) Slog.d(TAG, String.format("RCV <- {%s}", event));
32.                     String[] tokens = event.split("
    ", 2);
33.                     try {
34.                         int code = Integer.parseInt(tokens[0]);
36.                         if (code >= ResponseCode.UnsolicitedInformational) {
37.                             mCallbackHandler.sendMessage(
38.                                     mCallbackHandler.obtainMessage(code, event));
39.                         } else {
40.                             try {
41.                                 mResponseQueue.put(event);
42.                             } catch (InterruptedException ex) {
43.                                 Slog.e(TAG, "Failed to put response onto queue", ex);
44.                             }
45.                         }
46.                     } catch (NumberFormatException nfe) {
47.                         Slog.w(TAG, String.format("Bad msg (%s)", event));
48.                     }
49.                     start = i + 1;
50.                 }
51.             }
53.             // We should end at the amount we read. If not, compact then
54.             // buffer and read again.
55.             if (start != count) {
56.                 final int remaining = BUFFER_SIZE - start;
57.                 System.arraycopy(buffer, start, buffer, 0, remaining);
58.                 start = remaining;
59.             } else {
60.                 start = 0;
61.             }
62.         }
63.     } catch (IOException ex) {
64.         Slog.e(TAG, "Communications error", ex);
65.         throw ex;
66.     } finally {
67.         synchronized (mDaemonLock) {
68.             if (mOutputStream != null) {
69.                 try {
70.                     mOutputStream.close();
71.                 } catch (IOException e) {
72.                     Slog.w(TAG, "Failed closing output stream", e);
73.                 }
74.                 mOutputStream = null;
75.             }
76.         }
78.         try {
79.             if (socket != null) {
80.                 socket.close();
81.             }
82.         } catch (IOException ex) {
83.             Slog.w(TAG, "Failed closing socket", ex);
84.         }
85.     }
86. }

sendCommandLocked接口

1. /**
2.   * Sends a command to the daemon with a single argument
3.   *
4.   * @param command The command to send to the daemon
5.   * @param argument The argument to send with the command (or null)
6.   */
7.  private
   void sendCommandLocked(String command, String argument)
8.          throws NativeDaemonConnectorException {
9.      if (command != null && command.indexOf('\0') >= 0) {
10.          throw
    new IllegalArgumentException("unexpected command: " + command);
11.      }
12.      if (argument != null && argument.indexOf('\0') >= 0) {
13.          throw
    new IllegalArgumentException("unexpected argument: " + argument);
14.      }
16.      if (LOCAL_LOGD) Slog.d(TAG, String.format("SND -> {%s} {%s}", command, argument));
17.      if (mOutputStream == null) {
18.          Slog.e(TAG, "No connection to daemon", new IllegalStateException());
19.          throw
    new NativeDaemonConnectorException("No output stream!");
20.      } else {
21.          StringBuilder builder = new StringBuilder(command);
22.          if (argument != null) {
23.              builder.append(argument);
24.          }
25.          builder.append('\0');
27.          try {
28.              mOutputStream.write(builder.toString().getBytes());
29.          } catch (IOException ex) {
30.              Slog.e(TAG, "IOException in sendCommand", ex);
31.          }
32.      }
33.  }