serf_url_async_fetcher.h

Go to the documentation of this file.

// Copyright 2010 Google Inc.

#ifndef PAGESPEED_SYSTEM_SERF_URL_ASYNC_FETCHER_H_
#define PAGESPEED_SYSTEM_SERF_URL_ASYNC_FETCHER_H_

#include <cstddef>
#include <vector>

#include "apr_network_io.h"
#include "net/instaweb/http/public/url_async_fetcher.h"
#include "pagespeed/kernel/base/basictypes.h"
#include "pagespeed/kernel/base/gtest_prod.h"
#include "pagespeed/kernel/base/pool.h"
#include "pagespeed/kernel/base/pool_element.h"
#include "pagespeed/kernel/base/string.h"
#include "pagespeed/kernel/base/string_util.h"
#include "pagespeed/kernel/base/thread_annotations.h"
#include "pagespeed/kernel/base/thread_system.h"
#include "pagespeed/kernel/http/response_headers_parser.h"

#include "third_party/serf/src/serf.h"

#ifndef SERF_HTTPS_FETCHING
#define SERF_HTTPS_FETCHING 1
#endif

struct apr_pool_t;
struct apr_uri_t;
struct serf_context_t;

namespace net_instaweb {

class AsyncFetch;
class MessageHandler;
class Statistics;
class SerfFetch;
class SerfThreadedFetcher;
class Timer;
class UpDownCounter;
class Variable;

struct SerfStats {
  static const char kSerfFetchRequestCount[];
  static const char kSerfFetchByteCount[];
  static const char kSerfFetchTimeDurationMs[];
  static const char kSerfFetchCancelCount[];
  static const char kSerfFetchActiveCount[];
  static const char kSerfFetchTimeoutCount[];
  static const char kSerfFetchFailureCount[];
  static const char kSerfFetchCertErrors[];
  static const char kSerfFetchReadCalls[];

  static const char kSerfFetchUltimateSuccess[];

  static const char kSerfFetchUltimateFailure[];

  static const char kSerfFetchLastCheckTimestampMs[];
};

enum class SerfCompletionResult {
  kClientCancel,
  kSuccess,
  kFailure
};

#define SERF_HTTPS_KEYWORDS \
  "enable,disable,allow_self_signed," \
  "allow_unknown_certificate_authority,allow_certificate_not_yet_valid"

class SerfUrlAsyncFetcher : public UrlAsyncFetcher {
 public:
  enum WaitChoice {
    kThreadedOnly,
    kMainlineOnly,
    kThreadedAndMainline
  };

  SerfUrlAsyncFetcher(const char* proxy, apr_pool_t* pool,
                      ThreadSystem* thread_system,
                      Statistics* statistics, Timer* timer, int64 timeout_ms,
                      MessageHandler* handler);
  SerfUrlAsyncFetcher(SerfUrlAsyncFetcher* parent, const char* proxy);
  virtual ~SerfUrlAsyncFetcher();

  static void InitStats(Statistics* statistics);

  virtual void ShutDown();

  virtual bool SupportsHttps() const;

  virtual void Fetch(const GoogleString& url,
                     MessageHandler* message_handler,
                     AsyncFetch* callback);
  int Poll(int64 max_wait_ms);

  bool WaitForActiveFetches(int64 max_milliseconds,
                            MessageHandler* message_handler,
                            WaitChoice wait_choice);

  void FetchComplete(SerfFetch* fetch);

  void ReportCompletedFetchStats(const SerfFetch* fetch);

  void ReportFetchSuccessStats(SerfCompletionResult result,
                               const ResponseHeaders* headers,
                               const SerfFetch* fetch);


  apr_pool_t* pool() const { return pool_; }

  void PrintActiveFetches(MessageHandler* handler) const;
  virtual int64 timeout_ms() { return timeout_ms_; }
  ThreadSystem* thread_system() { return thread_system_; }

  void set_list_outstanding_urls_on_error(bool x);

  bool track_original_content_length() const {
    return track_original_content_length_;
  }
  void set_track_original_content_length(bool x);

  bool SetHttpsOptions(StringPiece directive);

  static bool ValidateHttpsOptions(StringPiece directive,
                                   GoogleString* error_message) {
    uint32 options;
    return ParseHttpsOptions(directive, &options, error_message);
  }

  void SetSslCertificatesDir(StringPiece dir);
  const GoogleString& ssl_certificates_dir() const {
    return ssl_certificates_dir_;
  }

  void SetSslCertificatesFile(StringPiece file);
  const GoogleString& ssl_certificates_file() const {
    return ssl_certificates_file_;
  }

 protected:
  typedef Pool<SerfFetch> SerfFetchPool;

  inline bool allow_https() const;
  inline bool allow_self_signed() const;
  inline bool allow_unknown_certificate_authority() const;
  inline bool allow_certificate_not_yet_valid() const;

  void set_https_options(uint32 https_options) {
    https_options_ = https_options;
  }

  void Init(apr_pool_t* parent_pool, const char* proxy)
      EXCLUSIVE_LOCKS_REQUIRED(mutex_);
  bool SetupProxy(const char* proxy) EXCLUSIVE_LOCKS_REQUIRED(mutex_);

  bool StartFetch(SerfFetch* fetch) EXCLUSIVE_LOCKS_REQUIRED(mutex_);

  virtual bool AnyPendingFetches();
  int ApproximateNumActiveFetches();

  void CancelActiveFetches();
  void CancelActiveFetchesMutexHeld() EXCLUSIVE_LOCKS_REQUIRED(mutex_);
  bool WaitForActiveFetchesHelper(int64 max_ms,
                                  MessageHandler* message_handler);

  void CleanupFetchesWithErrors() EXCLUSIVE_LOCKS_REQUIRED(mutex_);

  bool shutdown() const EXCLUSIVE_LOCKS_REQUIRED(mutex_) { return shutdown_; }
  void set_shutdown(bool s) EXCLUSIVE_LOCKS_REQUIRED(mutex_) { shutdown_ = s; }

  apr_pool_t* pool_;
  ThreadSystem* thread_system_;
  Timer* timer_;

  ThreadSystem::CondvarCapableMutex* mutex_;

  typedef std::vector<SerfFetch*> FetchVector;
  SerfFetchPool completed_fetches_;
  SerfThreadedFetcher* threaded_fetcher_;

  UpDownCounter* active_count_;

 private:
  friend class SerfFetch; 

  static const char* ExtractHostHeader(const apr_uri_t& uri,
                                       apr_pool_t* pool);
  FRIEND_TEST(SerfUrlAsyncFetcherTest, TestHostConstruction);

  static GoogleString RemovePortFromHostHeader(const GoogleString& in);
  FRIEND_TEST(SerfUrlAsyncFetcherTest, TestPortRemoval);

  static bool ParseHttpsOptions(StringPiece directive, uint32* options,
                                GoogleString* error_message);

  serf_context_t* serf_context_ GUARDED_BY(mutex_);
  SerfFetchPool active_fetches_ GUARDED_BY(mutex_);

  Variable* request_count_;
  Variable* byte_count_;
  Variable* time_duration_ms_;
  Variable* cancel_count_;
  Variable* timeout_count_;
  Variable* failure_count_;
  Variable* cert_errors_;
  Variable* read_calls_count_; 
  Variable* ultimate_success_;
  Variable* ultimate_failure_;
  UpDownCounter* last_check_timestamp_ms_;
  const int64 timeout_ms_;
  bool shutdown_ GUARDED_BY(mutex_);
  bool list_outstanding_urls_on_error_;
  bool track_original_content_length_;
  uint32 https_options_; 
  MessageHandler* message_handler_;
  GoogleString ssl_certificates_dir_;
  GoogleString ssl_certificates_file_;

  
};

class SerfFetch : public PoolElement<SerfFetch> {
 public:
  enum class CancelCause {
    kClientDecision,
    kSerfError,
    kFetchTimeout,
  };

  SerfFetch(const GoogleString& url,
            AsyncFetch* async_fetch,
            MessageHandler* message_handler,
            Timer* timer);
  ~SerfFetch();

  bool Start(SerfUrlAsyncFetcher* fetcher, serf_context_t* context);

  GoogleString DebugInfo();

  void Cancel(CancelCause cause);

  void CallCallback(SerfCompletionResult result);
  void CallbackDone(SerfCompletionResult result);

  void CleanupIfError();

  void ParseUrlForTesting(bool* status,
                          apr_uri_t** url,
                          const char** host_header,
                          const char** sni_host);

  void SetFetcherForTesting(SerfUrlAsyncFetcher* fetcher);

  int64 TimeDuration() const;

  int64 fetch_start_ms() const { return fetch_start_ms_; }

  size_t bytes_received() const { return bytes_received_; }
  MessageHandler* message_handler() { return message_handler_; }

 private:

#if SERF_HTTPS_FETCHING
  static apr_status_t SSLCertValidate(void *data, int failures,
                                   const serf_ssl_certificate_t *cert);

  static apr_status_t SSLCertChainValidate(
      void *data, int failures, int error_depth,
      const serf_ssl_certificate_t * const *certs,
      apr_size_t certs_count);
#endif

  static apr_status_t ConnectionSetup(
      apr_socket_t* socket, serf_bucket_t **read_bkt, serf_bucket_t **write_bkt,
      void* setup_baton, apr_pool_t* pool);
  static void ClosedConnection(serf_connection_t* conn,
                               void* closed_baton,
                               apr_status_t why,
                               apr_pool_t* pool);
  static serf_bucket_t* AcceptResponse(serf_request_t* request,
                                       serf_bucket_t* stream,
                                       void* acceptor_baton,
                                       apr_pool_t* pool);
  static apr_status_t HandleResponse(serf_request_t* request,
                                     serf_bucket_t* response,
                                     void* handler_baton,
                                     apr_pool_t* pool);
  static bool StatusIndicatesDataPossible(apr_status_t status);

#if SERF_HTTPS_FETCHING
  apr_status_t HandleSSLCertValidation(
      int errors, int failure_depth, const serf_ssl_certificate_t *cert);
#endif

  apr_status_t HandleResponse(serf_bucket_t* response);

  apr_status_t ReadStatusLine(serf_bucket_t* response);

  apr_status_t ReadHeaders(serf_bucket_t* response);

  apr_status_t ReadBody(serf_bucket_t* response);

  void FixUserAgent();
  static apr_status_t SetupRequest(serf_request_t* request,
                                   void* setup_baton,
                                   serf_bucket_t** req_bkt,
                                   serf_response_acceptor_t* acceptor,
                                   void** acceptor_baton,
                                   serf_response_handler_t* handler,
                                   void** handler_baton,
                                   apr_pool_t* pool);
  bool ParseUrl();

  SerfUrlAsyncFetcher* fetcher_;
  Timer* timer_;
  const GoogleString str_url_;
  AsyncFetch* async_fetch_;
  ResponseHeadersParser parser_;
  bool status_line_read_;
  MessageHandler* message_handler_;

  apr_pool_t* pool_;
  serf_bucket_alloc_t* bucket_alloc_;
  apr_uri_t url_;
  const char* host_header_; 
  const char* sni_host_; 
  serf_connection_t* connection_;
  size_t bytes_received_;
  int64 fetch_start_ms_;
  int64 fetch_end_ms_;

  bool using_https_;
  serf_ssl_context_t* ssl_context_;
  const char* ssl_error_message_;

  
};

}  

#endif