func (c *controllerAPI) AppLog(ctx context.Context, w http.ResponseWriter, req *http.Request) {
	ctx, cancel := context.WithCancel(ctx)

	opts := logaggc.LogOpts{
		Follow: req.FormValue("follow") == "true",
		JobID:  req.FormValue("job_id"),
	}
	if vals, ok := req.Form["process_type"]; ok && len(vals) > 0 {
		opts.ProcessType = &vals[len(vals)-1]
	}
	if strLines := req.FormValue("lines"); strLines != "" {
		lines, err := strconv.Atoi(req.FormValue("lines"))
		if err != nil {
			respondWithError(w, err)
			return
		}
		opts.Lines = &lines
	}
	rc, err := c.logaggc.GetLog(c.getApp(ctx).ID, &opts)
	if err != nil {
		respondWithError(w, err)
		return
	}

	if cn, ok := w.(http.CloseNotifier); ok {
		go func() {
			select {
			case <-cn.CloseNotify():
				rc.Close()
			case <-ctx.Done():
			}
		}()
	}
	defer cancel()
	defer rc.Close()

	if !strings.Contains(req.Header.Get("Accept"), "text/event-stream") {
		w.Header().Set("Content-Type", "text/plain")
		w.WriteHeader(200)
		// Send headers right away if following
		if wf, ok := w.(http.Flusher); ok && opts.Follow {
			wf.Flush()
		}

		fw := httphelper.FlushWriter{Writer: w, Enabled: opts.Follow}
		io.Copy(fw, rc)
		return
	}

	ch := make(chan *sseLogChunk)
	l, _ := ctxhelper.LoggerFromContext(ctx)
	s := sse.NewStream(w, ch, l)
	defer s.Close()
	s.Serve()

	msgc := make(chan *json.RawMessage)
	go func() {
		defer close(msgc)
		dec := json.NewDecoder(rc)
		for {
			var m json.RawMessage
			if err := dec.Decode(&m); err != nil {
				if err != io.EOF {
					l.Error("decoding logagg stream", err)
				}
				return
			}
			msgc <- &m
		}
	}()

	for {
		select {
		case m := <-msgc:
			if m == nil {
				ch <- &sseLogChunk{Event: "eof"}
				return
			}
			// write to sse
			select {
			case ch <- &sseLogChunk{Event: "message", Data: *m}:
			case <-s.Done:
				return
			case <-ctx.Done():
				return
			}
		case <-s.Done:
			return
		case <-ctx.Done():
			return
		}
	}
}

func (s *LogAggregatorTestSuite) TestAPIGetLogBuffer(c *C) {
	appID := "test-app"
	msg1 := newMessageForApp(appID, "web.1", "log message 1")
	msg2 := newMessageForApp(appID, "web.2", "log message 2")
	msg3 := newMessageForApp(appID, "worker.3", "log message 3")
	msg4 := newMessageForApp(appID, "web.1", "log message 4")
	msg5 := newMessageForApp(appID, ".5", "log message 5")
	buf := s.agg.getOrInitializeBuffer(appID)
	buf.Add(msg1)
	buf.Add(msg2)
	buf.Add(msg3)
	buf.Add(msg4)
	buf.Add(msg5)

	runtest := func(opts client.LogOpts, expected string) {
		numLines := -1
		if opts.Lines != nil {
			numLines = *opts.Lines
		}
		processType := "<nil>"
		if opts.ProcessType != nil {
			processType = *opts.ProcessType
		}
		c.Logf("Follow=%t Lines=%d JobID=%q ProcessType=%q", opts.Follow, numLines, opts.JobID, processType)
		logrc, err := s.client.GetLog(appID, &opts)
		c.Assert(err, IsNil)
		defer logrc.Close()

		assertAllLogsEquals(c, logrc, expected)
	}

	tests := []struct {
		numLogs     *int
		jobID       string
		processType *string
		expected    []*rfc5424.Message
	}{
		{
			numLogs:  intPtr(-1),
			expected: []*rfc5424.Message{msg1, msg2, msg3, msg4, msg5},
		},
		{
			numLogs:  intPtr(0),
			expected: nil,
		},
		{
			numLogs:  intPtr(1),
			expected: []*rfc5424.Message{msg5},
		},
		{
			numLogs:  intPtr(1),
			jobID:    "3",
			expected: []*rfc5424.Message{msg3},
		},
		{
			numLogs:  intPtr(-1),
			jobID:    "1",
			expected: []*rfc5424.Message{msg1, msg4},
		},
		{
			numLogs:     intPtr(-1),
			processType: strPtr("web"),
			expected:    []*rfc5424.Message{msg1, msg2, msg4},
		},
		{
			numLogs:     intPtr(-1),
			processType: strPtr(""),
			expected:    []*rfc5424.Message{msg5},
		},
	}
	for _, test := range tests {
		opts := client.LogOpts{
			Follow: false,
			JobID:  test.jobID,
		}
		if test.processType != nil {
			opts.ProcessType = test.processType
		}
		if test.numLogs != nil {
			opts.Lines = test.numLogs
		}
		expected := ""
		for _, msg := range test.expected {
			expected += marshalMessage(msg)
		}
		runtest(opts, expected)
	}
}