| 6382 | |
| 6383 | #define PIPEMODE_WRITE_LOOP_MAX_BYTES (128*1024) |
| 6384 | static void pipeMode(void) { |
| 6385 | long long errors = 0, replies = 0, obuf_len = 0, obuf_pos = 0; |
| 6386 | char obuf[1024*16]; /* Output buffer */ |
| 6387 | char aneterr[ANET_ERR_LEN]; |
| 6388 | redisReply *reply; |
| 6389 | int eof = 0; /* True once we consumed all the standard input. */ |
| 6390 | int done = 0; |
| 6391 | char magic[20]; /* Special reply we recognize. */ |
| 6392 | time_t last_read_time = time(NULL); |
| 6393 | |
| 6394 | srand(time(NULL)); |
| 6395 | |
| 6396 | /* Use non blocking I/O. */ |
| 6397 | if (anetNonBlock(aneterr,context->fd) == ANET_ERR) { |
| 6398 | fprintf(stderr, "Can't set the socket in non blocking mode: %s\n", |
| 6399 | aneterr); |
| 6400 | exit(1); |
| 6401 | } |
| 6402 | |
| 6403 | context->flags &= ~REDIS_BLOCK; |
| 6404 | |
| 6405 | /* Transfer raw protocol and read replies from the server at the same |
| 6406 | * time. */ |
| 6407 | while(!done) { |
| 6408 | int mask = AE_READABLE; |
| 6409 | |
| 6410 | if (!eof || obuf_len != 0) mask |= AE_WRITABLE; |
| 6411 | mask = aeWait(context->fd,mask,1000); |
| 6412 | |
| 6413 | /* Handle the readable state: we can read replies from the server. */ |
| 6414 | if (mask & AE_READABLE) { |
| 6415 | int read_error = 0; |
| 6416 | |
| 6417 | do { |
| 6418 | if (!read_error && redisBufferRead(context) == REDIS_ERR) { |
| 6419 | read_error = 1; |
| 6420 | } |
| 6421 | |
| 6422 | reply = NULL; |
| 6423 | if (redisGetReply(context, (void **) &reply) == REDIS_ERR) { |
| 6424 | fprintf(stderr, "Error reading replies from server\n"); |
| 6425 | exit(1); |
| 6426 | } |
| 6427 | if (reply) { |
| 6428 | last_read_time = time(NULL); |
| 6429 | if (reply->type == REDIS_REPLY_ERROR) { |
| 6430 | fprintf(stderr,"%s\n", reply->str); |
| 6431 | errors++; |
| 6432 | } else if (eof && reply->type == REDIS_REPLY_STRING && |
| 6433 | reply->len == 20) { |
| 6434 | /* Check if this is the reply to our final ECHO |
| 6435 | * command. If so everything was received |
| 6436 | * from the server. */ |
| 6437 | if (memcmp(reply->str,magic,20) == 0) { |
| 6438 | printf("Last reply received from server.\n"); |
| 6439 | done = 1; |
| 6440 | replies--; |
| 6441 | } |
no test coverage detected