| 694 | } |
| 695 | |
| 696 | static int dbq_poll(Lua L, dbconsumer_t *q, int delay_ms) |
| 697 | { |
| 698 | SP sp = getsp(L); |
| 699 | while (1) { |
| 700 | if (stop_waiting(L, q)) { |
| 701 | return -1; |
| 702 | } |
| 703 | int rc; |
| 704 | uint8_t status; |
| 705 | struct timespec ts; |
| 706 | Pthread_mutex_lock(q->lock); |
| 707 | again: status = *q->status; |
| 708 | if (status == TRIGGER_SUBSCRIPTION_OPEN) { |
| 709 | rc = dbq_poll_int(L, q); // call will release q->lock |
| 710 | } else if (status == TRIGGER_SUBSCRIPTION_PAUSED) { |
| 711 | if (stop_waiting(L, q)) { |
| 712 | return -1; |
| 713 | } |
| 714 | ts = setup_dbq_ts(delay_ms); |
| 715 | pthread_cond_timedwait(q->cond, q->lock, &ts); /* RC IGNORED */ |
| 716 | goto again; |
| 717 | } else { |
| 718 | assert(status == TRIGGER_SUBSCRIPTION_CLOSED); |
| 719 | Pthread_mutex_unlock(q->lock); |
| 720 | rc = -2; |
| 721 | } |
| 722 | if (rc == 1) { |
| 723 | return rc; |
| 724 | } |
| 725 | if (rc < 0) { |
| 726 | luabb_error(L, sp, "failed to read from:%s rc:%d", q->info.spname, rc); |
| 727 | return rc; |
| 728 | } |
| 729 | if (delay_ms <= 0) { |
| 730 | return 0; |
| 731 | } |
| 732 | ts = setup_dbq_ts(delay_ms); |
| 733 | Pthread_mutex_lock(q->lock); |
| 734 | if (pthread_cond_timedwait(q->cond, q->lock, &ts) == 0) { |
| 735 | // was woken up -- try getting from queue |
| 736 | goto again; |
| 737 | } |
| 738 | Pthread_mutex_unlock(q->lock); |
| 739 | delay_ms -= dbq_delay_ms; |
| 740 | if (delay_ms <= 0) { |
| 741 | return 0; |
| 742 | } |
| 743 | } |
| 744 | } |
| 745 | |
| 746 | // this call will block until queue item available |
| 747 | static int dbconsumer_get_int(Lua L, dbconsumer_t *q) |
no test coverage detected