// transformResponse converts an API response into a structured API object.
func (r *Request) transformResponse(resp *http.Response, req *http.Request) ([]byte, bool, error) {
	defer resp.Body.Close()

	body, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return nil, false, err
	}

	// Did the server give us a status response?
	isStatusResponse := false
	var status api.Status
	if err := r.codec.DecodeInto(body, &status); err == nil && status.Status != "" {
		isStatusResponse = true
	}

	switch {
	case resp.StatusCode < http.StatusOK || resp.StatusCode > http.StatusPartialContent:
		if !isStatusResponse {
			return nil, false, fmt.Errorf("request [%#v] failed (%d) %s: %s", req, resp.StatusCode, resp.Status, string(body))
		}
		return nil, false, errors.FromObject(&status)
	}

	// If the server gave us a status back, look at what it was.
	if isStatusResponse && status.Status != api.StatusSuccess {
		// "Working" requests need to be handled specially.
		// "Failed" requests are clearly just an error and it makes sense to return them as such.
		return nil, false, errors.FromObject(&status)
	}

	created := resp.StatusCode == http.StatusCreated
	return body, created, err
}

// transformResponse converts an API response into a structured API object.
func (r *Request) transformResponse(resp *http.Response, req *http.Request) ([]byte, bool, error) {
	defer resp.Body.Close()

	body, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return nil, false, err
	}

	// Did the server give us a status response?
	isStatusResponse := false
	var status api.Status
	if err := r.codec.DecodeInto(body, &status); err == nil && status.Status != "" {
		isStatusResponse = true
	}

	switch {
	case resp.StatusCode == http.StatusSwitchingProtocols:
		// no-op, we've been upgraded
	case resp.StatusCode < http.StatusOK || resp.StatusCode > http.StatusPartialContent:
		if !isStatusResponse {
			var err error = &UnexpectedStatusError{
				Request:  req,
				Response: resp,
				Body:     string(body),
			}
			// TODO: handle other error classes we know about
			switch resp.StatusCode {
			case http.StatusConflict:
				if req.Method == "POST" {
					err = errors.NewAlreadyExists(r.resource, r.resourceName)
				} else {
					err = errors.NewConflict(r.resource, r.resourceName, err)
				}
			case http.StatusNotFound:
				err = errors.NewNotFound(r.resource, r.resourceName)
			case http.StatusBadRequest:
				err = errors.NewBadRequest(err.Error())
			}
			return nil, false, err
		}
		return nil, false, errors.FromObject(&status)
	}

	// If the server gave us a status back, look at what it was.
	if isStatusResponse && status.Status != api.StatusSuccess {
		// "Working" requests need to be handled specially.
		// "Failed" requests are clearly just an error and it makes sense to return them as such.
		return nil, false, errors.FromObject(&status)
	}

	created := resp.StatusCode == http.StatusCreated
	return body, created, err
}

// finishRequest makes a given resultFunc asynchronous and handles errors returned by the response.
// Any api.Status object returned is considered an "error", which interrupts the normal response flow.
func finishRequest(timeout time.Duration, fn resultFunc) (result runtime.Object, err error) {
	// these channels need to be buffered to prevent the goroutine below from hanging indefinitely
	// when the select statement reads something other than the one the goroutine sends on.
	ch := make(chan runtime.Object, 1)
	errCh := make(chan error, 1)
	go func() {
		if result, err := fn(); err != nil {
			errCh <- err
		} else {
			ch <- result
		}
	}()

	select {
	case result = <-ch:
		if status, ok := result.(*api.Status); ok {
			return nil, errors.FromObject(status)
		}
		return result, nil
	case err = <-errCh:
		return nil, err
	case <-time.After(timeout):
		return nil, errors.NewTimeoutError("request did not complete within allowed duration", 0)
	}
}

// watchHandler watches w and keeps *resourceVersion up to date.
func (r *Reflector) watchHandler(w watch.Interface, resourceVersion *string, resyncCh <-chan time.Time, stopCh <-chan struct{}) error {
	start := time.Now()
	eventCount := 0

	// Stopping the watcher should be idempotent and if we return from this function there's no way
	// we're coming back in with the same watch interface.
	defer w.Stop()

loop:
	for {
		select {
		case <-stopCh:
			return errorStopRequested
		case <-resyncCh:
			return errorResyncRequested
		case event, ok := <-w.ResultChan():
			if !ok {
				break loop
			}
			if event.Type == watch.Error {
				return apierrs.FromObject(event.Object)
			}
			if e, a := r.expectedType, reflect.TypeOf(event.Object); e != a {
				util.HandleError(fmt.Errorf("%s: expected type %v, but watch event object had type %v", r.name, e, a))
				continue
			}
			meta, err := meta.Accessor(event.Object)
			if err != nil {
				util.HandleError(fmt.Errorf("%s: unable to understand watch event %#v", r.name, event))
				continue
			}
			switch event.Type {
			case watch.Added:
				r.store.Add(event.Object)
			case watch.Modified:
				r.store.Update(event.Object)
			case watch.Deleted:
				// TODO: Will any consumers need access to the "last known
				// state", which is passed in event.Object? If so, may need
				// to change this.
				r.store.Delete(event.Object)
			default:
				util.HandleError(fmt.Errorf("%s: unable to understand watch event %#v", r.name, event))
			}
			*resourceVersion = meta.ResourceVersion()
			r.setLastSyncResourceVersion(*resourceVersion)
			eventCount++
		}
	}

	watchDuration := time.Now().Sub(start)
	if watchDuration < 1*time.Second && eventCount == 0 {
		glog.V(4).Infof("%s: Unexpected watch close - watch lasted less than a second and no items received", r.name)
		return errors.New("very short watch")
	}
	glog.V(4).Infof("%s: Watch close - %v total %v items received", r.name, r.expectedType, eventCount)
	return nil
}

// resourceVersion is a pointer to the resource version to use/update.
func (rm *ReplicationManager) watchControllers(resourceVersion *string) {
	watching, err := rm.kubeClient.ReplicationControllers(api.NamespaceAll).Watch(
		labels.Everything(),
		fields.Everything(),
		*resourceVersion,
	)
	if err != nil {
		util.HandleError(fmt.Errorf("unable to watch: %v", err))
		time.Sleep(5 * time.Second)
		return
	}

	for {
		select {
		case <-rm.syncTime:
			rm.synchronize()
		case event, open := <-watching.ResultChan():
			if !open {
				// watchChannel has been closed, or something else went
				// wrong with our watch call. Let the util.Forever()
				// that called us call us again.
				return
			}
			if event.Type == watch.Error {
				util.HandleError(fmt.Errorf("error from watch during sync: %v", errors.FromObject(event.Object)))
				// Clear the resource version, this may cause us to skip some elements on the watch,
				// but we'll catch them on the synchronize() call, so it works out.
				*resourceVersion = ""
				continue
			}
			glog.V(4).Infof("Got watch: %#v", event)
			rc, ok := event.Object.(*api.ReplicationController)
			if !ok {
				if status, ok := event.Object.(*api.Status); ok {
					if status.Status == api.StatusFailure {
						glog.Errorf("failed to watch: %v", status)
						// Clear resource version here, as above, this won't hurt consistency, but we
						// should consider introspecting more carefully here. (or make the apiserver smarter)
						// "why not both?"
						*resourceVersion = ""
						continue
					}
				}
				util.HandleError(fmt.Errorf("unexpected object: %#v", event.Object))
				continue
			}
			// If we get disconnected, start where we left off.
			*resourceVersion = rc.ResourceVersion
			// Sync even if this is a deletion event, to ensure that we leave
			// it in the desired state.
			glog.V(4).Infof("About to sync from watch: %v", rc.Name)
			if err := rm.syncHandler(*rc); err != nil {
				util.HandleError(fmt.Errorf("unexpected sync error: %v", err))
			}
		}
	}
}

// transformResponse converts an API response into a structured API object
func (r *Request) transformResponse(resp *http.Response, req *http.Request) Result {
	var body []byte
	if resp.Body != nil {
		if data, err := ioutil.ReadAll(resp.Body); err == nil {
			body = data
		}
	}
	glog.V(8).Infof("Response Body: %s", string(body))

	// Did the server give us a status response?
	isStatusResponse := false
	var status api.Status
	if err := r.codec.DecodeInto(body, &status); err == nil && status.Status != "" {
		isStatusResponse = true
	}

	switch {
	case resp.StatusCode == http.StatusSwitchingProtocols:
		// no-op, we've been upgraded
	case resp.StatusCode < http.StatusOK || resp.StatusCode > http.StatusPartialContent:
		if !isStatusResponse {
			return Result{err: r.transformUnstructuredResponseError(resp, req, body)}
		}
		return Result{err: errors.FromObject(&status)}
	}

	// If the server gave us a status back, look at what it was.
	success := resp.StatusCode >= http.StatusOK && resp.StatusCode <= http.StatusPartialContent
	if isStatusResponse && (status.Status != api.StatusSuccess && !success) {
		// "Failed" requests are clearly just an error and it makes sense to return them as such.
		return Result{err: errors.FromObject(&status)}
	}

	return Result{
		body:    body,
		created: resp.StatusCode == http.StatusCreated,
		codec:   r.codec,
	}
}

// watchHandler watches w and keeps *resourceVersion up to date.
func (r *Reflector) watchHandler(w watch.Interface, resourceVersion *string, exitWatch <-chan time.Time) error {
	start := time.Now()
	eventCount := 0
loop:
	for {
		select {
		case <-exitWatch:
			w.Stop()
			return errorResyncRequested
		case event, ok := <-w.ResultChan():
			if !ok {
				break loop
			}
			if event.Type == watch.Error {
				return apierrs.FromObject(event.Object)
			}
			if e, a := r.expectedType, reflect.TypeOf(event.Object); e != a {
				glog.Errorf("expected type %v, but watch event object had type %v", e, a)
				continue
			}
			meta, err := meta.Accessor(event.Object)
			if err != nil {
				glog.Errorf("unable to understand watch event %#v", event)
				continue
			}
			switch event.Type {
			case watch.Added:
				r.store.Add(event.Object)
			case watch.Modified:
				r.store.Update(event.Object)
			case watch.Deleted:
				// TODO: Will any consumers need access to the "last known
				// state", which is passed in event.Object? If so, may need
				// to change this.
				r.store.Delete(event.Object)
			default:
				glog.Errorf("unable to understand watch event %#v", event)
			}
			*resourceVersion = meta.ResourceVersion()
			eventCount++
		}
	}

	watchDuration := time.Now().Sub(start)
	if watchDuration < 1*time.Second && eventCount == 0 {
		glog.V(4).Infof("Unexpected watch close - watch lasted less than a second and no items received")
		return errors.New("very short watch")
	}
	glog.V(4).Infof("Watch close - %v total %v items received", r.expectedType, eventCount)
	return nil
}

// Stream formats and executes the request, and offers streaming of the response.
// Returns io.ReadCloser which could be used for streaming of the response, or an error
// Any non-2xx http status code causes an error.  If we get a non-2xx code, we try to convert the body into an APIStatus object.
// If we can, we return that as an error.  Otherwise, we create an error that lists the http status and the content of the response.
func (r *Request) Stream() (io.ReadCloser, error) {
	if r.err != nil {
		return nil, r.err
	}
	url := r.URL().String()
	req, err := http.NewRequest(r.verb, url, nil)
	if err != nil {
		return nil, err
	}
	client := r.client
	if client == nil {
		client = http.DefaultClient
	}
	resp, err := client.Do(req)
	if err != nil {
		return nil, err
	}

	switch {
	case (resp.StatusCode >= 200) && (resp.StatusCode < 300):
		return resp.Body, nil

	default:
		// ensure we close the body before returning the error
		defer resp.Body.Close()

		// we have a decent shot at taking the object returned, parsing it as a status object and returning a more normal error
		bodyBytes, err := ioutil.ReadAll(resp.Body)
		if err != nil {
			return nil, fmt.Errorf("%v while accessing %v", resp.Status, url)
		}

		if runtimeObject, err := r.codec.Decode(bodyBytes); err == nil {
			statusError := errors.FromObject(runtimeObject)

			if _, ok := statusError.(APIStatus); ok {
				return nil, statusError
			}
		}

		bodyText := string(bodyBytes)
		return nil, fmt.Errorf("%s while accessing %v: %s", resp.Status, url, bodyText)
	}

	return resp.Body, nil
}

// resourceVersion is a pointer to the resource version to use/update.
func (rm *ReplicationManager) watchControllers(resourceVersion *string) {
	watching, err := rm.kubeClient.ReplicationControllers(api.NamespaceAll).Watch(
		labels.Everything(),
		labels.Everything(),
		*resourceVersion,
	)
	if err != nil {
		glog.Errorf("Unexpected failure to watch: %v", err)
		time.Sleep(5 * time.Second)
		return
	}

	for {
		select {
		case <-rm.syncTime:
			rm.synchronize()
		case event, open := <-watching.ResultChan():
			if !open {
				// watchChannel has been closed, or something else went
				// wrong with our etcd watch call. Let the util.Forever()
				// that called us call us again.
				return
			}
			if event.Type == watch.Error {
				glog.Errorf("error from watch during sync: %v", errors.FromObject(event.Object))
				continue
			}
			glog.V(4).Infof("Got watch: %#v", event)
			rc, ok := event.Object.(*api.ReplicationController)
			if !ok {
				glog.Errorf("unexpected object: %#v", event.Object)
				continue
			}
			// If we get disconnected, start where we left off.
			*resourceVersion = rc.ResourceVersion
			// Sync even if this is a deletion event, to ensure that we leave
			// it in the desired state.
			glog.V(4).Infof("About to sync from watch: %v", rc.Name)
			if err := rm.syncHandler(*rc); err != nil {
				glog.Errorf("unexpected sync. error: %v", err)
			}
		}
	}
}