| 7292 | |
| 7293 | #define PIPEMODE_WRITE_LOOP_MAX_BYTES (128*1024) |
| 7294 | static void pipeMode(void) { |
| 7295 | long long errors = 0, replies = 0, obuf_len = 0, obuf_pos = 0; |
| 7296 | char obuf[1024*16]; /* Output buffer */ |
| 7297 | char aneterr[ANET_ERR_LEN]; |
| 7298 | redisReply *reply; |
| 7299 | int eof = 0; /* True once we consumed all the standard input. */ |
| 7300 | int done = 0; |
| 7301 | char magic[20]; /* Special reply we recognize. */ |
| 7302 | time_t last_read_time = time(NULL); |
| 7303 | |
| 7304 | srand(time(NULL)); |
| 7305 | |
| 7306 | /* Use non blocking I/O. */ |
| 7307 | if (anetNonBlock(aneterr,context->fd) == ANET_ERR) { |
| 7308 | fprintf(stderr, "Can't set the socket in non blocking mode: %s\n", |
| 7309 | aneterr); |
| 7310 | exit(1); |
| 7311 | } |
| 7312 | |
| 7313 | context->flags &= ~REDIS_BLOCK; |
| 7314 | |
| 7315 | /* Transfer raw protocol and read replies from the server at the same |
| 7316 | * time. */ |
| 7317 | while(!done) { |
| 7318 | int mask = AE_READABLE; |
| 7319 | |
| 7320 | if (!eof || obuf_len != 0) mask |= AE_WRITABLE; |
| 7321 | mask = aeWait(context->fd,mask,1000); |
| 7322 | |
| 7323 | /* Handle the readable state: we can read replies from the server. */ |
| 7324 | if (mask & AE_READABLE) { |
| 7325 | int read_error = 0; |
| 7326 | |
| 7327 | do { |
| 7328 | if (!read_error && redisBufferRead(context) == REDIS_ERR) { |
| 7329 | read_error = 1; |
| 7330 | } |
| 7331 | |
| 7332 | reply = NULL; |
| 7333 | if (redisGetReply(context, (void **) &reply) == REDIS_ERR) { |
| 7334 | fprintf(stderr, "Error reading replies from server\n"); |
| 7335 | exit(1); |
| 7336 | } |
| 7337 | if (reply) { |
| 7338 | last_read_time = time(NULL); |
| 7339 | if (reply->type == REDIS_REPLY_ERROR) { |
| 7340 | fprintf(stderr,"%s\n", reply->str); |
| 7341 | errors++; |
| 7342 | } else if (eof && reply->type == REDIS_REPLY_STRING && |
| 7343 | reply->len == 20) { |
| 7344 | /* Check if this is the reply to our final ECHO |
| 7345 | * command. If so everything was received |
| 7346 | * from the server. */ |
| 7347 | if (memcmp(reply->str,magic,20) == 0) { |
| 7348 | printf("Last reply received from server.\n"); |
| 7349 | done = 1; |
| 7350 | replies--; |
| 7351 | } |
no test coverage detected