`Server` implementation.
| 2005 | |
| 2006 | // `Server` implementation. |
| 2007 | Future<Nothing> run() |
| 2008 | { |
| 2009 | return state.transition<State::INITIALIZED, State::RUNNING>([=]() { |
| 2010 | // Start the accept loop and store the future so we can later |
| 2011 | // discard it when we need to stop the server. |
| 2012 | accepting = loop( |
| 2013 | self(), |
| 2014 | [=]() { |
| 2015 | return socket.accept(); |
| 2016 | }, |
| 2017 | [=](const network::Socket& socket) -> ControlFlow<Nothing> { |
| 2018 | // If we've transitioned to STOPPING we should break. It |
| 2019 | // may seem like we should never get here because we |
| 2020 | // discard the accept loop before we transition to |
| 2021 | // STOPPING but it's possible that we've already |
| 2022 | // dispatched this lambda and it is only now getting |
| 2023 | // invoked. It's critical that we break because after we |
| 2024 | // transition to STOPPING we assume that `clients` will |
| 2025 | // not be modified by the accept loop. |
| 2026 | if (state.is<State::STOPPING>()) { |
| 2027 | return Break(); |
| 2028 | } |
| 2029 | |
| 2030 | Client client = { |
| 2031 | /* .socket = */ socket, |
| 2032 | /* .serving = */ http::serve( |
| 2033 | socket, |
| 2034 | [=](const Request& request) { |
| 2035 | return f(socket, request); |
| 2036 | }) |
| 2037 | }; |
| 2038 | |
| 2039 | clients.put(socket, client); |
| 2040 | |
| 2041 | client.serving |
| 2042 | .onAny(defer(self(), [=](const Future<Nothing>&) { |
| 2043 | clients.erase(socket); |
| 2044 | })); |
| 2045 | |
| 2046 | return Continue(); |
| 2047 | }); |
| 2048 | |
| 2049 | // We return a _discardable_ `accepting` so the caller can stop |
| 2050 | // running a server by "discarding the run", for example: |
| 2051 | // |
| 2052 | // Future<Nothing> run = server.run(); |
| 2053 | // run.discard(); |
| 2054 | // |
| 2055 | // Even if we returned an _undiscardable_ `accepting` we still |
| 2056 | // need to do a `recover()` on it in the event that the accept |
| 2057 | // loop fails (or is abandoned) so we can stop the server (if it |
| 2058 | // isn't already being stopped). If `accepting` completes |
| 2059 | // successfully then we must be stopping so just wait until |
| 2060 | // we've stopped! |
| 2061 | return accepting |
| 2062 | .then(defer(self(), [=]() { |
| 2063 | return state.when<State::STOPPED>(); |
| 2064 | })) |