| 153 | } |
| 154 | |
| 155 | int IOChannel::TransMsgDirect(int source_fd, int dest_fd, struct pollfd pf[], int nfds) { |
| 156 | int byte_size = 0; |
| 157 | int read_once = 0; |
| 158 | int written = 0; |
| 159 | |
| 160 | for (int i = 0; i < nfds; ++i) { |
| 161 | char buf[1024 * 16]; |
| 162 | if (pf[i].fd == source_fd) { |
| 163 | if (pf[i].revents & POLLIN) { |
| 164 | if ((read_once = read(source_fd, buf, sizeof(buf))) <= 0) { |
| 165 | if (read_once < 0) { |
| 166 | MonitorPluginEntry::GetDefault()->GetMonitorPlugin()->ReadNetworkFail(); |
| 167 | } |
| 168 | LOG_ERR("clientfd %d svrfd %d source %d dest %d read failed, ret %d, errno (%d:%s)", |
| 169 | client_fd_, sqlsvr_fd_, source_fd, dest_fd, read_once, errno, strerror(errno)); |
| 170 | return -1; |
| 171 | } |
| 172 | |
| 173 | if (source_fd == client_fd_) { |
| 174 | if (!is_authed_) { |
| 175 | GetDBNameFromAuthBuf(buf, read_once); |
| 176 | is_authed_ = true; |
| 177 | } else { |
| 178 | GetDBNameFromReqBuf(buf, read_once); |
| 179 | } |
| 180 | |
| 181 | if (!CanExecute(buf, read_once)) { |
| 182 | LOG_ERR("request [cmd:%d] can't execute here", (int) buf[4]); |
| 183 | return -1; |
| 184 | } |
| 185 | } |
| 186 | |
| 187 | written = WriteToDest(dest_fd, buf, read_once); |
| 188 | if (written < 0) { |
| 189 | return -__LINE__; |
| 190 | } |
| 191 | |
| 192 | byte_size += written; |
| 193 | //LOG_DEBUG("last read clientfd %d svrfd %d source %d dest %d ret %d read ret %d", |
| 194 | // client_fd_, sqlsvr_fd_, source_fd, dest_fd, read_once, byte_size); |
| 195 | } else if (pf[i].revents & POLLHUP) { |
| 196 | LOG_DEBUG("source %d dest %d read events POLLHUP", source_fd, dest_fd); |
| 197 | } else if (pf[i].revents & POLLERR) { |
| 198 | return -__LINE__; |
| 199 | } |
| 200 | } |
| 201 | } |
| 202 | return byte_size; |
| 203 | } |
| 204 | |
| 205 | int IOChannel::FakeClientIPInAuthBuf(char * buf, size_t buf_len) { |
| 206 | if (buf_len < 36) { |
nothing calls this directly
no test coverage detected