| 791 | } |
| 792 | |
| 793 | void _read(const Pipe::Reader& reader, const Future<Result<Event>>& event) |
| 794 | { |
| 795 | CHECK(!event.isDiscarded()); |
| 796 | |
| 797 | // Ignore enqueued events from the previous Subscribe call reader. |
| 798 | if (!subscribed.isSome() || subscribed->reader != reader) { |
| 799 | VLOG(1) << "Ignoring event from old stale connection"; |
| 800 | return; |
| 801 | } |
| 802 | |
| 803 | CHECK_EQ(SUBSCRIBED, state); |
| 804 | CHECK_SOME(connectionId); |
| 805 | |
| 806 | // This could happen if the master failed over while sending a event. |
| 807 | if (event.isFailed()) { |
| 808 | LOG(ERROR) << "Failed to decode the stream of events: " |
| 809 | << event.failure(); |
| 810 | disconnected(connectionId.get(), event.failure()); |
| 811 | return; |
| 812 | } |
| 813 | |
| 814 | // This could happen if the master failed over after sending an event. |
| 815 | if (event->isNone()) { |
| 816 | const string error = "End-Of-File received from master. The master " |
| 817 | "closed the event stream"; |
| 818 | LOG(ERROR) << error; |
| 819 | |
| 820 | disconnected(connectionId.get(), error); |
| 821 | return; |
| 822 | } |
| 823 | |
| 824 | if (event->isError()) { |
| 825 | error("Failed to de-serialize event: " + event->error()); |
| 826 | } else { |
| 827 | receive(event->get(), false); |
| 828 | } |
| 829 | |
| 830 | read(); |
| 831 | } |
| 832 | |
| 833 | void receive(const Event& event, bool isLocallyInjected) |
| 834 | { |