| 739 | } |
| 740 | |
| 741 | void _read(const Pipe::Reader& reader, const Future<Result<Event>>& event) |
| 742 | { |
| 743 | CHECK(!event.isDiscarded()); |
| 744 | |
| 745 | // Ignore enqueued events from the previous Subscribe call reader. |
| 746 | if (subscribed.isNone() || subscribed->reader != reader) { |
| 747 | VLOG(1) << "Ignoring event from old stale connection"; |
| 748 | return; |
| 749 | } |
| 750 | |
| 751 | CHECK_EQ(SUBSCRIBED, state); |
| 752 | CHECK_SOME(connectionId); |
| 753 | |
| 754 | // This could happen if the agent process died while sending a response. |
| 755 | if (event.isFailed()) { |
| 756 | LOG(ERROR) << "Failed to decode the stream of events: " |
| 757 | << event.failure(); |
| 758 | |
| 759 | disconnected(connectionId.get(), event.failure()); |
| 760 | return; |
| 761 | } |
| 762 | |
| 763 | // This could happen if the agent failed over after sending an event. |
| 764 | if (event->isNone()) { |
| 765 | const string error = "End-Of-File received from agent. The agent closed" |
| 766 | " the event stream"; |
| 767 | LOG(ERROR) << error; |
| 768 | |
| 769 | disconnected(connectionId.get(), error); |
| 770 | return; |
| 771 | } |
| 772 | |
| 773 | if (event->isError()) { |
| 774 | error("Failed to de-serialize event: " + event->error()); |
| 775 | return; |
| 776 | } |
| 777 | |
| 778 | receive(event->get(), false); |
| 779 | read(); |
| 780 | } |
| 781 | |
| 782 | void receive(const Event& event, bool isLocallyInjected) |
| 783 | { |