func TestLoadBufferedEventsOnlyFromPast(t *testing.T) {
	now := time.Now()
	f, err := timetypes.GetTimestamp("2016-03-07T17:28:03.090000000+02:00", now)
	if err != nil {
		t.Fatal(err)
	}
	s, sNano, err := timetypes.ParseTimestamps(f, 0)
	if err != nil {
		t.Fatal(err)
	}

	f, err = timetypes.GetTimestamp("2016-03-07T17:28:03.100000000+02:00", now)
	if err != nil {
		t.Fatal(err)
	}
	u, uNano, err := timetypes.ParseTimestamps(f, 0)
	if err != nil {
		t.Fatal(err)
	}

	m1, err := eventstestutils.Scan("2016-03-07T17:28:03.022433271+02:00 container die 0b863f2a26c18557fc6cdadda007c459f9ec81b874780808138aea78a3595079 (image=ubuntu, name=small_hoover)")
	if err != nil {
		t.Fatal(err)
	}
	m2, err := eventstestutils.Scan("2016-03-07T17:28:03.091719377+02:00 network disconnect 19c5ed41acb798f26b751e0035cd7821741ab79e2bbd59a66b5fd8abf954eaa0 (type=bridge, container=0b863f2a26c18557fc6cdadda007c459f9ec81b874780808138aea78a3595079, name=bridge)")
	if err != nil {
		t.Fatal(err)
	}
	m3, err := eventstestutils.Scan("2016-03-07T17:28:03.129014751+02:00 container destroy 0b863f2a26c18557fc6cdadda007c459f9ec81b874780808138aea78a3595079 (image=ubuntu, name=small_hoover)")
	if err != nil {
		t.Fatal(err)
	}

	events := &Events{
		events: []events.Message{*m1, *m2, *m3},
	}

	since := time.Unix(s, sNano)
	until := time.Unix(u, uNano)

	out := events.loadBufferedEvents(since, until, nil)
	if len(out) != 1 {
		t.Fatalf("expected 1 message, got %d: %v", len(out), out)
	}

	if out[0].Type != "network" {
		t.Fatalf("expected network event, got %s", out[0].Type)
	}
}

// Scan turns an event string like the default ones formatted in the cli output
// and turns it into an event message.
func Scan(text string) (*events.Message, error) {
	md := ScanMap(text)
	if len(md) == 0 {
		return nil, fmt.Errorf("text is not an event: %s", text)
	}

	f, err := timetypes.GetTimestamp(md["timestamp"], time.Now())
	if err != nil {
		return nil, err
	}

	t, tn, err := timetypes.ParseTimestamps(f, -1)
	if err != nil {
		return nil, err
	}

	attrs := make(map[string]string)
	for _, a := range strings.SplitN(md["attributes"], ", ", -1) {
		kv := strings.SplitN(a, "=", 2)
		attrs[kv[0]] = kv[1]
	}

	tu := time.Unix(t, tn)
	return &events.Message{
		Time:     t,
		TimeNano: tu.UnixNano(),
		Type:     md["eventType"],
		Action:   md["action"],
		Actor: events.Actor{
			ID:         md["id"],
			Attributes: attrs,
		},
	}, nil
}

func eventTime(formTime string) (time.Time, error) {
	t, tNano, err := timetypes.ParseTimestamps(formTime, -1)
	if err != nil {
		return time.Time{}, err
	}
	if t == -1 {
		return time.Time{}, nil
	}
	return time.Unix(t, tNano), nil
}

func getUntilFromPruneFilters(pruneFilters filters.Args) (time.Time, error) {
	until := time.Time{}
	if !pruneFilters.Include("until") {
		return until, nil
	}
	untilFilters := pruneFilters.Get("until")
	if len(untilFilters) > 1 {
		return until, fmt.Errorf("more than one until filter specified")
	}
	ts, err := timetypes.GetTimestamp(untilFilters[0], time.Now())
	if err != nil {
		return until, err
	}
	seconds, nanoseconds, err := timetypes.ParseTimestamps(ts, 0)
	if err != nil {
		return until, err
	}
	until = time.Unix(seconds, nanoseconds)
	return until, nil
}

func (s *systemRouter) getEvents(ctx context.Context, w http.ResponseWriter, r *http.Request, vars map[string]string) error {
	if err := httputils.ParseForm(r); err != nil {
		return err
	}
	since, sinceNano, err := timetypes.ParseTimestamps(r.Form.Get("since"), -1)
	if err != nil {
		return err
	}
	until, untilNano, err := timetypes.ParseTimestamps(r.Form.Get("until"), -1)
	if err != nil {
		return err
	}

	timer := time.NewTimer(0)
	timer.Stop()
	if until > 0 || untilNano > 0 {
		dur := time.Unix(until, untilNano).Sub(time.Now())
		timer = time.NewTimer(dur)
	}

	ef, err := filters.FromParam(r.Form.Get("filters"))
	if err != nil {
		return err
	}

	w.Header().Set("Content-Type", "application/json")

	// This is to ensure that the HTTP status code is sent immediately,
	// so that it will not block the receiver.
	w.WriteHeader(http.StatusOK)
	if flusher, ok := w.(http.Flusher); ok {
		flusher.Flush()
	}

	output := ioutils.NewWriteFlusher(w)
	defer output.Close()

	enc := json.NewEncoder(output)

	buffered, l := s.backend.SubscribeToEvents(since, sinceNano, ef)
	defer s.backend.UnsubscribeFromEvents(l)

	for _, ev := range buffered {
		if err := enc.Encode(ev); err != nil {
			return err
		}
	}

	var closeNotify <-chan bool
	if closeNotifier, ok := w.(http.CloseNotifier); ok {
		closeNotify = closeNotifier.CloseNotify()
	}

	for {
		select {
		case ev := <-l:
			jev, ok := ev.(*jsonmessage.JSONMessage)
			if !ok {
				continue
			}
			if err := enc.Encode(jev); err != nil {
				return err
			}
		case <-timer.C:
			return nil
		case <-closeNotify:
			logrus.Debug("Client disconnected, stop sending events")
			return nil
		}
	}
}

// ContainerLogs hooks up a container's stdout and stderr streams
// configured with the given struct.
func (daemon *Daemon) ContainerLogs(ctx context.Context, containerName string, config *backend.ContainerLogsConfig, started chan struct{}) error {
	container, err := daemon.GetContainer(containerName)
	if err != nil {
		return err
	}

	if !(config.ShowStdout || config.ShowStderr) {
		return fmt.Errorf("You must choose at least one stream")
	}

	cLog, err := daemon.getLogger(container)
	if err != nil {
		return err
	}
	logReader, ok := cLog.(logger.LogReader)
	if !ok {
		return logger.ErrReadLogsNotSupported
	}

	follow := config.Follow && container.IsRunning()
	tailLines, err := strconv.Atoi(config.Tail)
	if err != nil {
		tailLines = -1
	}

	logrus.Debug("logs: begin stream")

	var since time.Time
	if config.Since != "" {
		s, n, err := timetypes.ParseTimestamps(config.Since, 0)
		if err != nil {
			return err
		}
		since = time.Unix(s, n)
	}
	readConfig := logger.ReadConfig{
		Since:  since,
		Tail:   tailLines,
		Follow: follow,
	}
	logs := logReader.ReadLogs(readConfig)

	wf := ioutils.NewWriteFlusher(config.OutStream)
	defer wf.Close()
	close(started)
	wf.Flush()

	var outStream io.Writer
	outStream = wf
	errStream := outStream
	if !container.Config.Tty {
		errStream = stdcopy.NewStdWriter(outStream, stdcopy.Stderr)
		outStream = stdcopy.NewStdWriter(outStream, stdcopy.Stdout)
	}

	for {
		select {
		case err := <-logs.Err:
			logrus.Errorf("Error streaming logs: %v", err)
			return nil
		case <-ctx.Done():
			logs.Close()
			return nil
		case msg, ok := <-logs.Msg:
			if !ok {
				logrus.Debug("logs: end stream")
				logs.Close()
				if cLog != container.LogDriver {
					// Since the logger isn't cached in the container, which occurs if it is running, it
					// must get explicitly closed here to avoid leaking it and any file handles it has.
					if err := cLog.Close(); err != nil {
						logrus.Errorf("Error closing logger: %v", err)
					}
				}
				return nil
			}
			logLine := msg.Line
			if config.Details {
				logLine = append([]byte(msg.Attrs.String()+" "), logLine...)
			}
			if config.Timestamps {
				logLine = append([]byte(msg.Timestamp.Format(logger.TimeFormat)+" "), logLine...)
			}
			if msg.Source == "stdout" && config.ShowStdout {
				outStream.Write(logLine)
			}
			if msg.Source == "stderr" && config.ShowStderr {
				errStream.Write(logLine)
			}
		}
	}
}

func (s *containerRouter) getContainersLogs(ctx context.Context, w http.ResponseWriter, r *http.Request, vars map[string]string) error {
	if err := httputils.ParseForm(r); err != nil {
		return err
	}

	// Args are validated before the stream starts because when it starts we're
	// sending HTTP 200 by writing an empty chunk of data to tell the client that
	// daemon is going to stream. By sending this initial HTTP 200 we can't report
	// any error after the stream starts (i.e. container not found, wrong parameters)
	// with the appropriate status code.
	stdout, stderr := httputils.BoolValue(r, "stdout"), httputils.BoolValue(r, "stderr")
	if !(stdout || stderr) {
		return fmt.Errorf("Bad parameters: you must choose at least one stream")
	}

	var since time.Time
	if r.Form.Get("since") != "" {
		s, n, err := timetypes.ParseTimestamps(r.Form.Get("since"), 0)
		if err != nil {
			return err
		}
		since = time.Unix(s, n)
	}

	var closeNotifier <-chan bool
	if notifier, ok := w.(http.CloseNotifier); ok {
		closeNotifier = notifier.CloseNotify()
	}

	containerName := vars["name"]

	if !s.backend.Exists(containerName) {
		return derr.ErrorCodeNoSuchContainer.WithArgs(containerName)
	}

	// write an empty chunk of data (this is to ensure that the
	// HTTP Response is sent immediately, even if the container has
	// not yet produced any data)
	w.WriteHeader(http.StatusOK)
	if flusher, ok := w.(http.Flusher); ok {
		flusher.Flush()
	}

	output := ioutils.NewWriteFlusher(w)
	defer output.Close()

	logsConfig := &daemon.ContainerLogsConfig{
		Follow:     httputils.BoolValue(r, "follow"),
		Timestamps: httputils.BoolValue(r, "timestamps"),
		Since:      since,
		Tail:       r.Form.Get("tail"),
		UseStdout:  stdout,
		UseStderr:  stderr,
		OutStream:  output,
		Stop:       closeNotifier,
	}

	if err := s.backend.ContainerLogs(containerName, logsConfig); err != nil {
		// The client may be expecting all of the data we're sending to
		// be multiplexed, so send it through OutStream, which will
		// have been set up to handle that if needed.
		fmt.Fprintf(logsConfig.OutStream, "Error running logs job: %s\n", utils.GetErrorMessage(err))
	}

	return nil
}