* Wait until we can read a CopyData message, * or timeout, or occurrence of a signal or input on the stop_socket. * (timeout_ms < 0 means wait indefinitely; 0 means don't wait.) * * Returns 1 if data has become available for reading, 0 if timed out * or interrupted by signal or stop_socket input, and -1 on an error. */
| 871 | * or interrupted by signal or stop_socket input, and -1 on an error. |
| 872 | */ |
| 873 | static int |
| 874 | CopyStreamPoll(PGconn *conn, long timeout_ms, pgsocket stop_socket) |
| 875 | { |
| 876 | int ret; |
| 877 | fd_set input_mask; |
| 878 | int connsocket; |
| 879 | int maxfd; |
| 880 | struct timeval timeout; |
| 881 | struct timeval *timeoutptr; |
| 882 | |
| 883 | connsocket = PQsocket(conn); |
| 884 | if (connsocket < 0) |
| 885 | { |
| 886 | pg_log_error("invalid socket: %s", PQerrorMessage(conn)); |
| 887 | return -1; |
| 888 | } |
| 889 | |
| 890 | FD_ZERO(&input_mask); |
| 891 | FD_SET(connsocket, &input_mask); |
| 892 | maxfd = connsocket; |
| 893 | if (stop_socket != PGINVALID_SOCKET) |
| 894 | { |
| 895 | FD_SET(stop_socket, &input_mask); |
| 896 | maxfd = Max(maxfd, stop_socket); |
| 897 | } |
| 898 | |
| 899 | if (timeout_ms < 0) |
| 900 | timeoutptr = NULL; |
| 901 | else |
| 902 | { |
| 903 | timeout.tv_sec = timeout_ms / 1000L; |
| 904 | timeout.tv_usec = (timeout_ms % 1000L) * 1000L; |
| 905 | timeoutptr = &timeout; |
| 906 | } |
| 907 | |
| 908 | ret = select(maxfd + 1, &input_mask, NULL, NULL, timeoutptr); |
| 909 | |
| 910 | if (ret < 0) |
| 911 | { |
| 912 | if (errno == EINTR) |
| 913 | return 0; /* Got a signal, so not an error */ |
| 914 | pg_log_error("%s() failed: %m", "select"); |
| 915 | return -1; |
| 916 | } |
| 917 | if (ret > 0 && FD_ISSET(connsocket, &input_mask)) |
| 918 | return 1; /* Got input on connection socket */ |
| 919 | |
| 920 | return 0; /* Got timeout or input on stop_socket */ |
| 921 | } |
| 922 | |
| 923 | /* |
| 924 | * Receive CopyData message available from XLOG stream, blocking for |
no test coverage detected