// handleWatch processes a watch request
func (h *WatchHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
	parts := splitPath(req.URL.Path)
	if len(parts) < 1 || req.Method != "GET" {
		notFound(w, req)
		return
	}
	storage := h.storage[parts[0]]
	if storage == nil {
		notFound(w, req)
		return
	}
	if watcher, ok := storage.(ResourceWatcher); ok {
		label, field, resourceVersion := getWatchParams(req.URL.Query())
		watching, err := watcher.Watch(label, field, resourceVersion)
		if err != nil {
			errorJSON(err, h.codec, w)
			return
		}

		// TODO: This is one watch per connection. We want to multiplex, so that
		// multiple watches of the same thing don't create two watches downstream.
		watchServer := &WatchServer{watching}
		if req.Header.Get("Connection") == "Upgrade" && req.Header.Get("Upgrade") == "websocket" {
			websocket.Handler(watchServer.HandleWS).ServeHTTP(httplog.Unlogged(w), req)
		} else {
			watchServer.ServeHTTP(w, req)
		}
		return
	}

	notFound(w, req)
}

// serveWatch handles serving requests to the server
func serveWatch(watcher watch.Interface, scope RequestScope, w http.ResponseWriter, req *restful.Request) {
	watchServer := &WatchServer{watcher, scope.Codec, func(obj runtime.Object) {
		if err := setSelfLink(obj, req, scope.Namer); err != nil {
			glog.V(5).Infof("Failed to set self link for object %v: %v", reflect.TypeOf(obj), err)
		}
	}}
	if isWebsocketRequest(req.Request) {
		websocket.Handler(watchServer.HandleWS).ServeHTTP(httplog.Unlogged(w), req.Request)
	} else {
		watchServer.ServeHTTP(w, req.Request)
	}
}

// serveWatch handles serving requests to the server
func serveWatch(watcher watch.Interface, scope RequestScope, w http.ResponseWriter, req *restful.Request) {
	// Each watch gets a random timeout to avoid thundering herds. Rand is seeded once in the api installer.
	timeout := time.Duration(MinTimeoutSecs+rand.Intn(MaxTimeoutSecs-MinTimeoutSecs)) * time.Second

	watchServer := &WatchServer{watcher, scope.Codec, func(obj runtime.Object) {
		if err := setSelfLink(obj, req, scope.Namer); err != nil {
			glog.V(5).Infof("Failed to set self link for object %v: %v", reflect.TypeOf(obj), err)
		}
	}, &realTimeoutFactory{timeout}}
	if isWebsocketRequest(req.Request) {
		websocket.Handler(watchServer.HandleWS).ServeHTTP(httplog.Unlogged(w), req.Request)
	} else {
		watchServer.ServeHTTP(w, req.Request)
	}
}

// ServeHTTP processes watch requests.
func (h *WatchHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
	if req.Method != "GET" {
		notFound(w, req)
		return
	}

	namespace, kind, _, err := KindAndNamespace(req)
	if err != nil {
		notFound(w, req)
		return
	}
	ctx := api.WithNamespace(api.NewContext(), namespace)

	storage := h.storage[kind]
	if storage == nil {
		notFound(w, req)
		return
	}
	watcher, ok := storage.(ResourceWatcher)
	if !ok {
		errorJSON(errors.NewMethodNotSupported(kind, "watch"), h.codec, w)
		return
	}

	label, field, resourceVersion, err := getWatchParams(req.URL.Query())
	if err != nil {
		errorJSON(err, h.codec, w)
		return
	}
	watching, err := watcher.Watch(ctx, label, field, resourceVersion)
	if err != nil {
		errorJSON(err, h.codec, w)
		return
	}

	// TODO: This is one watch per connection. We want to multiplex, so that
	// multiple watches of the same thing don't create two watches downstream.
	watchServer := &WatchServer{watching, h.codec, func(obj runtime.Object) {
		if err := h.setSelfLinkAddName(obj, req); err != nil {
			glog.Errorf("Failed to set self link for object %#v", obj)
		}
	}}
	if isWebsocketRequest(req) {
		websocket.Handler(watchServer.HandleWS).ServeHTTP(httplog.Unlogged(w), req)
	} else {
		watchServer.ServeHTTP(w, req)
	}
}

// ServeHTTP serves a series of JSON encoded events via straight HTTP with
// Transfer-Encoding: chunked.
func (self *WatchServer) ServeHTTP(w http.ResponseWriter, req *http.Request) {
	loggedW := httplog.LogOf(req, w)
	w = httplog.Unlogged(w)

	cn, ok := w.(http.CloseNotifier)
	if !ok {
		loggedW.Addf("unable to get CloseNotifier")
		http.NotFound(w, req)
		return
	}
	flusher, ok := w.(http.Flusher)
	if !ok {
		loggedW.Addf("unable to get Flusher")
		http.NotFound(w, req)
		return
	}

	w.Header().Set("Transfer-Encoding", "chunked")
	w.WriteHeader(http.StatusOK)
	flusher.Flush()

	encoder := json.NewEncoder(w)
	for {
		select {
		case <-cn.CloseNotify():
			self.watching.Stop()
			return
		case event, ok := <-self.watching.ResultChan():
			if !ok {
				// End of results.
				return
			}
			obj, err := api.NewJSONWatchEvent(self.codec, event)
			if err != nil {
				// Client disconnect.
				self.watching.Stop()
				return
			}
			if err := encoder.Encode(obj); err != nil {
				// Client disconnect.
				self.watching.Stop()
				return
			}
			flusher.Flush()
		}
	}
}

// serveWatch handles serving requests to the server
func serveWatch(watcher watch.Interface, scope RequestScope, w http.ResponseWriter, req *restful.Request, minRequestTimeout time.Duration) {
	var timeout time.Duration
	if minRequestTimeout > 0 {
		// Each watch gets a random timeout between minRequestTimeout and 2*minRequestTimeout to avoid thundering herds.
		timeout = time.Duration(float64(minRequestTimeout) * (rand.Float64() + 1.0))
	}
	watchServer := &WatchServer{watcher, scope.Codec, func(obj runtime.Object) {
		if err := setSelfLink(obj, req, scope.Namer); err != nil {
			glog.V(5).Infof("Failed to set self link for object %v: %v", reflect.TypeOf(obj), err)
		}
	}, &realTimeoutFactory{timeout}}
	if isWebsocketRequest(req.Request) {
		websocket.Handler(watchServer.HandleWS).ServeHTTP(httplog.Unlogged(w), req.Request)
	} else {
		watchServer.ServeHTTP(w, req.Request)
	}
}

// ServeHTTP serves a series of JSON encoded events via straight HTTP with
// Transfer-Encoding: chunked.
func (self *WatchServer) ServeHTTP(w http.ResponseWriter, req *http.Request) {
	loggedW := httplog.LogOf(req, w)
	w = httplog.Unlogged(w)
	timeoutCh, cleanup := self.t.TimeoutCh()
	defer cleanup()
	defer self.watching.Stop()

	cn, ok := w.(http.CloseNotifier)
	if !ok {
		loggedW.Addf("unable to get CloseNotifier")
		http.NotFound(w, req)
		return
	}
	flusher, ok := w.(http.Flusher)
	if !ok {
		loggedW.Addf("unable to get Flusher")
		http.NotFound(w, req)
		return
	}
	w.Header().Set("Transfer-Encoding", "chunked")
	w.WriteHeader(http.StatusOK)
	flusher.Flush()
	encoder := watchjson.NewEncoder(w, self.codec)
	for {
		select {
		case <-cn.CloseNotify():
			return
		case <-timeoutCh:
			return
		case event, ok := <-self.watching.ResultChan():
			if !ok {
				// End of results.
				return
			}
			self.fixup(event.Object)
			if err := encoder.Encode(&event); err != nil {
				// Client disconnect.
				return
			}
			flusher.Flush()
		}
	}
}

// ServeHTTP processes watch requests.
func (h *WatchHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
	ctx := api.NewContext()
	namespace := req.URL.Query().Get("namespace")
	if len(namespace) > 0 {
		ctx = api.WithNamespace(ctx, namespace)
	}
	parts := splitPath(req.URL.Path)
	if len(parts) < 1 || req.Method != "GET" {
		notFound(w, req)
		return
	}
	storage := h.storage[parts[0]]
	if storage == nil {
		notFound(w, req)
		return
	}
	if watcher, ok := storage.(ResourceWatcher); ok {
		label, field, resourceVersion := getWatchParams(req.URL.Query())
		watching, err := watcher.Watch(ctx, label, field, resourceVersion)
		if err != nil {
			errorJSON(err, h.codec, w)
			return
		}

		// TODO: This is one watch per connection. We want to multiplex, so that
		// multiple watches of the same thing don't create two watches downstream.
		watchServer := &WatchServer{watching, h.codec, func(obj runtime.Object) {
			if err := h.setSelfLinkAddName(obj, req); err != nil {
				glog.Errorf("Failed to set self link for object %#v", obj)
			}
		}}
		if isWebsocketRequest(req) {
			websocket.Handler(watchServer.HandleWS).ServeHTTP(httplog.Unlogged(w), req)
		} else {
			watchServer.ServeHTTP(w, req)
		}
		return
	}

	notFound(w, req)
}

func (server *APIServer) handleWatch(w http.ResponseWriter, req *http.Request) {
	prefix := server.watchPrefix()
	if !strings.HasPrefix(req.URL.Path, prefix) {
		server.notFound(w, req)
		return
	}
	parts := strings.Split(req.URL.Path[len(prefix):], "/")[1:]
	if req.Method != "GET" || len(parts) < 1 {
		server.notFound(w, req)
	}
	storage := server.storage[parts[0]]
	if storage == nil {
		server.notFound(w, req)
	}
	if watcher, ok := storage.(ResourceWatcher); ok {
		var watching watch.Interface
		var err error
		if id := req.URL.Query().Get("id"); id != "" {
			watching, err = watcher.WatchSingle(id)
		} else {
			watching, err = watcher.WatchAll()
		}
		if err != nil {
			server.error(err, w)
			return
		}

		// TODO: This is one watch per connection. We want to multiplex, so that
		// multiple watches of the same thing don't create two watches downstream.
		watchServer := &WatchServer{watching}
		if req.Header.Get("Connection") == "Upgrade" && req.Header.Get("Upgrade") == "websocket" {
			websocket.Handler(watchServer.HandleWS).ServeHTTP(httplog.Unlogged(w), req)
		} else {
			watchServer.ServeHTTP(w, req)
		}
		return
	}

	server.notFound(w, req)
}

// ServeHTTP processes watch requests.
func (h *WatchHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
	var verb string
	var apiResource string
	var httpCode int
	reqStart := time.Now()
	defer monitor("watch", &verb, &apiResource, &httpCode, reqStart)

	if req.Method != "GET" {
		httpCode = errorJSON(errors.NewBadRequest(
			fmt.Sprintf("unsupported method for watch: %s", req.Method)), h.codec, w)
		return
	}

	requestInfo, err := h.info.GetAPIRequestInfo(req)
	if err != nil {
		httpCode = errorJSON(errors.NewBadRequest(
			fmt.Sprintf("failed to find api request info: %s", err.Error())), h.codec, w)
		return
	}
	verb = requestInfo.Verb
	ctx := api.WithNamespace(api.NewContext(), requestInfo.Namespace)

	storage := h.storage[requestInfo.Resource]
	if storage == nil {
		httpCode = errorJSON(errors.NewNotFound(requestInfo.Resource, "Resource"), h.codec, w)
		return
	}
	apiResource = requestInfo.Resource
	watcher, ok := storage.(ResourceWatcher)
	if !ok {
		httpCode = errorJSON(errors.NewMethodNotSupported(requestInfo.Resource, "watch"), h.codec, w)
		return
	}

	label, field, err := parseSelectorQueryParams(req.URL.Query(), requestInfo.APIVersion, apiResource)
	if err != nil {
		httpCode = errorJSON(err, h.codec, w)
		return
	}

	resourceVersion := req.URL.Query().Get("resourceVersion")
	watching, err := watcher.Watch(ctx, label, field, resourceVersion)
	if err != nil {
		httpCode = errorJSON(err, h.codec, w)
		return
	}
	httpCode = http.StatusOK

	// TODO: This is one watch per connection. We want to multiplex, so that
	// multiple watches of the same thing don't create two watches downstream.
	watchServer := &WatchServer{watching, h.codec, func(obj runtime.Object) {
		if err := h.setSelfLinkAddName(obj, req); err != nil {
			glog.Errorf("Failed to set self link for object %#v", obj)
		}
	}}
	if isWebsocketRequest(req) {
		websocket.Handler(watchServer.HandleWS).ServeHTTP(httplog.Unlogged(w), req)
	} else {
		watchServer.ServeHTTP(w, req)
	}
}