OLD | NEW |
(Empty) | |
| 1 // Copyright 2014 The Chromium Authors. All rights reserved. |
| 2 // Use of this source code is governed by a BSD-style license that can be |
| 3 // found in the LICENSE file. |
| 4 |
| 5 #include "athena/system/device_socket_listener.h" |
| 6 |
| 7 #include <errno.h> |
| 8 #include <map> |
| 9 #include <sys/socket.h> |
| 10 #include <sys/types.h> |
| 11 #include <vector> |
| 12 |
| 13 #include "base/bind.h" |
| 14 #include "base/files/file_path.h" |
| 15 #include "base/memory/singleton.h" |
| 16 #include "base/message_loop/message_loop.h" |
| 17 #include "base/observer_list.h" |
| 18 #include "base/stl_util.h" |
| 19 #include "content/public/browser/browser_thread.h" |
| 20 #include "ipc/unix_domain_socket_util.h" |
| 21 |
| 22 namespace athena { |
| 23 |
| 24 namespace { |
| 25 |
| 26 typedef ObserverList<DeviceSocketListener> DeviceSocketListeners; |
| 27 |
| 28 // Reads from a device socket blocks of a particular size. When that amount of |
| 29 // data is read DeviceSocketManager::OnDataAvailable is called on the singleton |
| 30 // instance which then informs all of the listeners on that socket. |
| 31 class DeviceSocketReader : public base::MessagePumpLibevent::Watcher { |
| 32 public: |
| 33 DeviceSocketReader(const std::string& socket_path, |
| 34 size_t data_size) |
| 35 : socket_path_(socket_path), |
| 36 data_size_(data_size), |
| 37 data_(new char[data_size]) { |
| 38 } |
| 39 virtual ~DeviceSocketReader() {} |
| 40 |
| 41 private: |
| 42 // Overidden from base::MessagePumpLibevent::Watcher. |
| 43 virtual void OnFileCanReadWithoutBlocking(int fd) OVERRIDE; |
| 44 virtual void OnFileCanWriteWithoutBlocking(int fd) OVERRIDE; |
| 45 |
| 46 std::string socket_path_; |
| 47 size_t data_size_; |
| 48 scoped_ptr<char[]> data_; |
| 49 |
| 50 DISALLOW_COPY_AND_ASSIGN(DeviceSocketReader); |
| 51 }; |
| 52 |
| 53 // A singleton instance for managing all connections to sockets. |
| 54 class DeviceSocketManager { |
| 55 public: |
| 56 static DeviceSocketManager* GetInstance() { |
| 57 return Singleton<DeviceSocketManager>::get(); |
| 58 } |
| 59 |
| 60 // If there isn't an existing connection to |socket_path|, then opens a |
| 61 // connection to |socket_path| and starts listening for data. All listeners |
| 62 // for |socket_path| receives data when data is available on the socket. |
| 63 void StartListening(const std::string& socket_path, |
| 64 size_t data_size, |
| 65 DeviceSocketListener* listener); |
| 66 |
| 67 // Removes |listener| from the list of listeners that receive data from |
| 68 // |socket_path|. If this is the last listener, then this closes the |
| 69 // connection to the socket. |
| 70 void StopListening(const std::string& socket_path, |
| 71 DeviceSocketListener* listener); |
| 72 |
| 73 // Sends data to all the listeners registered to receive data from |
| 74 // |socket_path|. |
| 75 void OnDataAvailable(const std::string& socket_path, |
| 76 const void* data); |
| 77 |
| 78 // Notifies listeners of errors reading from the socket and closes it. |
| 79 void OnError(const std::string& socket_path, int err); |
| 80 void OnEOF(const std::string& socket_path); |
| 81 |
| 82 private: |
| 83 friend struct DefaultSingletonTraits<DeviceSocketManager>; |
| 84 |
| 85 struct SocketData { |
| 86 SocketData() |
| 87 : fd(-1) { |
| 88 } |
| 89 |
| 90 int fd; |
| 91 DeviceSocketListeners observers; |
| 92 scoped_ptr<base::MessagePumpLibevent::FileDescriptorWatcher> controller; |
| 93 scoped_ptr<DeviceSocketReader> watcher; |
| 94 }; |
| 95 |
| 96 DeviceSocketManager() { |
| 97 } |
| 98 |
| 99 ~DeviceSocketManager() { |
| 100 STLDeleteContainerPairSecondPointers(socket_data_.begin(), |
| 101 socket_data_.end()); |
| 102 } |
| 103 |
| 104 void StartListeningOnIO(const std::string& socket_path, |
| 105 size_t data_size, |
| 106 DeviceSocketListener* listener); |
| 107 |
| 108 void StopListeningOnIO(const std::string& socket_path, |
| 109 DeviceSocketListener* listener); |
| 110 |
| 111 void CloseSocket(const std::string& socket_path); |
| 112 |
| 113 std::map<std::string, SocketData*> socket_data_; |
| 114 |
| 115 DISALLOW_COPY_AND_ASSIGN(DeviceSocketManager); |
| 116 }; |
| 117 |
| 118 //////////////////////////////////////////////////////////////////////////////// |
| 119 // DeviceSocketReader |
| 120 |
| 121 void DeviceSocketReader::OnFileCanReadWithoutBlocking(int fd) { |
| 122 ssize_t read_size = recv(fd, data_.get(), data_size_, 0); |
| 123 if (read_size < 0) { |
| 124 if (errno == EINTR) |
| 125 return; |
| 126 DeviceSocketManager::GetInstance()->OnError(socket_path_, errno); |
| 127 return; |
| 128 } |
| 129 if (read_size == 0) { |
| 130 DeviceSocketManager::GetInstance()->OnEOF(socket_path_); |
| 131 return; |
| 132 } |
| 133 if (read_size != static_cast<ssize_t>(data_size_)) |
| 134 return; |
| 135 DeviceSocketManager::GetInstance()->OnDataAvailable(socket_path_, |
| 136 data_.get()); |
| 137 } |
| 138 |
| 139 void DeviceSocketReader::OnFileCanWriteWithoutBlocking(int fd) { |
| 140 NOTREACHED(); |
| 141 } |
| 142 |
| 143 //////////////////////////////////////////////////////////////////////////////// |
| 144 // DeviceSocketManager |
| 145 |
| 146 void DeviceSocketManager::StartListening(const std::string& socket_path, |
| 147 size_t data_size, |
| 148 DeviceSocketListener* listener) { |
| 149 content::BrowserThread::PostTask(content::BrowserThread::IO, FROM_HERE, |
| 150 base::Bind(&DeviceSocketManager::StartListeningOnIO, |
| 151 base::Unretained(this), socket_path, data_size, listener)); |
| 152 } |
| 153 |
| 154 void DeviceSocketManager::StopListening(const std::string& socket_path, |
| 155 DeviceSocketListener* listener) { |
| 156 content::BrowserThread::PostTask(content::BrowserThread::IO, FROM_HERE, |
| 157 base::Bind(&DeviceSocketManager::StopListeningOnIO, |
| 158 base::Unretained(this), socket_path, listener)); |
| 159 } |
| 160 |
| 161 void DeviceSocketManager::OnDataAvailable(const std::string& socket_path, |
| 162 const void* data) { |
| 163 CHECK_GT(socket_data_.count(socket_path), 0UL); |
| 164 DeviceSocketListeners& listeners = socket_data_[socket_path]->observers; |
| 165 FOR_EACH_OBSERVER(DeviceSocketListener, listeners, OnDataAvailableOnIO(data)); |
| 166 } |
| 167 |
| 168 void DeviceSocketManager::CloseSocket(const std::string& socket_path) { |
| 169 if (!socket_data_.count(socket_path)) |
| 170 return; |
| 171 SocketData* socket_data = socket_data_[socket_path]; |
| 172 close(socket_data->fd); |
| 173 delete socket_data; |
| 174 socket_data_.erase(socket_path); |
| 175 } |
| 176 |
| 177 void DeviceSocketManager::OnError(const std::string& socket_path, int err) { |
| 178 LOG(ERROR) << "Error reading from socket: " << socket_path << ": " |
| 179 << strerror(err); |
| 180 CloseSocket(socket_path); |
| 181 // TODO(flackr): Notify listeners that the socket was closed unexpectedly. |
| 182 } |
| 183 |
| 184 void DeviceSocketManager::OnEOF(const std::string& socket_path) { |
| 185 LOG(ERROR) << "EOF reading from socket: " << socket_path; |
| 186 CloseSocket(socket_path); |
| 187 } |
| 188 |
| 189 void DeviceSocketManager::StartListeningOnIO(const std::string& socket_path, |
| 190 size_t data_size, |
| 191 DeviceSocketListener* listener) { |
| 192 CHECK(content::BrowserThread::CurrentlyOn(content::BrowserThread::IO)); |
| 193 SocketData* socket_data = NULL; |
| 194 if (!socket_data_.count(socket_path)) { |
| 195 int socket_fd = -1; |
| 196 if (!IPC::CreateClientUnixDomainSocket(base::FilePath(socket_path), |
| 197 &socket_fd)) { |
| 198 LOG(ERROR) << "Error connecting to socket: " << socket_path; |
| 199 return; |
| 200 } |
| 201 |
| 202 socket_data = new SocketData; |
| 203 socket_data_[socket_path] = socket_data; |
| 204 |
| 205 socket_data->fd = socket_fd; |
| 206 |
| 207 socket_data->controller.reset( |
| 208 new base::MessagePumpLibevent::FileDescriptorWatcher()); |
| 209 socket_data->watcher.reset( |
| 210 new DeviceSocketReader(socket_path, data_size)); |
| 211 |
| 212 base::MessageLoopForIO::current()->WatchFileDescriptor( |
| 213 socket_fd, |
| 214 true, |
| 215 base::MessageLoopForIO::WATCH_READ, |
| 216 socket_data->controller.get(), |
| 217 socket_data->watcher.get()); |
| 218 } else { |
| 219 socket_data = socket_data_[socket_path]; |
| 220 } |
| 221 socket_data->observers.AddObserver(listener); |
| 222 } |
| 223 |
| 224 void DeviceSocketManager::StopListeningOnIO(const std::string& socket_path, |
| 225 DeviceSocketListener* listener) { |
| 226 if (!socket_data_.count(socket_path)) |
| 227 return; // Happens if unable to create a socket. |
| 228 |
| 229 CHECK(content::BrowserThread::CurrentlyOn(content::BrowserThread::IO)); |
| 230 DeviceSocketListeners& listeners = socket_data_[socket_path]->observers; |
| 231 listeners.RemoveObserver(listener); |
| 232 if (!listeners.might_have_observers()) { |
| 233 // All listeners for this socket has been removed. Close the socket. |
| 234 CloseSocket(socket_path); |
| 235 } |
| 236 } |
| 237 |
| 238 } // namespace |
| 239 |
| 240 DeviceSocketListener::DeviceSocketListener(const std::string& socket_path, |
| 241 size_t data_size) |
| 242 : socket_path_(socket_path), |
| 243 data_size_(data_size) { |
| 244 } |
| 245 |
| 246 DeviceSocketListener::~DeviceSocketListener() { |
| 247 StopListening(); |
| 248 } |
| 249 |
| 250 void DeviceSocketListener::StartListening() { |
| 251 DeviceSocketManager::GetInstance()->StartListening(socket_path_, |
| 252 data_size_, |
| 253 this); |
| 254 } |
| 255 |
| 256 void DeviceSocketListener::StopListening() { |
| 257 DeviceSocketManager::GetInstance()->StopListening(socket_path_, this); |
| 258 } |
| 259 |
| 260 } // namespace athena |
OLD | NEW |