<commit_before>// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at http://mozilla.org/MPL/2.0/.

package transport

import (
	"encoding/json"
	"io"
	"io/ioutil"
	"log"
	"net/http"
	"strconv"
	"time"

	"github.com/julienschmidt/httprouter"
	"golang.org/x/net/context"

	"github.com/ninibe/bigduration"
	"github.com/ninibe/netlog"
)

// NewHTTPTransport transport sets up an HTTP interface around a NetLog.
func NewHTTPTransport(nl *netlog.NetLog) *HTTPTransport {
	return &HTTPTransport{nl: nl}
}

// HTTPTransport implements an HTTP server around a NetLog.
type HTTPTransport struct {
	nl *netlog.NetLog
}

// ServeHTTP implements the http.Handler interface around a NetLog.
func (ht *HTTPTransport) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	router := httprouter.New()
	router.GET("/", ht.handleServerInfo)
	router.GET("/:topic", ht.handleTopicInfo)
	router.POST("/:topic", ht.handleCreateTopic)
	router.POST("/:topic/payload", ht.handleWritePayload)
	router.GET("/:topic/payload/:offset", ht.handleReadPayload)
	router.GET("/:topic/sync", ht.handleSync)
	router.POST("/:topic/scanner", ht.handleCreateScanner)
	//router.DELETE("/:topic/scanner", ht.handleDeleteScanner)
	router.GET("/:topic/scan", withCtx(ht.handleScanTopic))
	router.DELETE("/:topic", ht.handleDeleteTopic)
	router.ServeHTTP(w, r)
	return
}

// ctxHandle is the signature a context-friendly http handler.
type ctxHandle func(context.Context, http.ResponseWriter, *http.Request, httprouter.Params)

// withCtx is a wrapper function to inject a context into an http handler
// the context gets canceled if the http connection is closed by the client.
func withCtx(handle ctxHandle) httprouter.Handle {
	ctx := context.Background()
	return func(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
		var cancel context.CancelFunc
		ctx, cancel = context.WithCancel(ctx)
		clientGone := w.(http.CloseNotifier).CloseNotify()
		go func() {
			select {
			case <-ctx.Done():
			case <-clientGone:
				cancel()
			}
		}()

		handle(ctx, w, r, ps)
	}
}

func (ht *HTTPTransport) handleCreateTopic(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {

	var settings netlog.TopicSettings
	encoder := json.NewDecoder(r.Body)
	err := encoder.Decode(&settings)
	if err != nil && err != io.EOF {
		log.Printf(err.Error())
		JSONErrorResponse(w, netlog.ErrBadRequest)
		return
	}

	_, err = ht.nl.CreateTopic(ps.ByName("topic"), settings)
	if err != nil {
		JSONErrorResponse(w, err)
		return
	}

	w.WriteHeader(http.StatusCreated)
	JSONOKResponse(w, "topic created")
}

func (ht *HTTPTransport) handleReadPayload(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
	t, err := ht.nl.Topic(ps.ByName("topic"))
	if err != nil {
		JSONErrorResponse(w, err)
		return
	}

	defer r.Body.Close()

	offset, err := t.ParseOffset(ps.ByName("offset"))
	if err != nil {
		JSONErrorResponse(w, netlog.ErrInvalidOffset)
		return
	}

	data, err := t.Payload(offset)
	if err != nil {
		JSONErrorResponse(w, err)
		return
	}

	w.Header().Set("Content-Type", "application/octet-stream")
	_, err = w.Write(data)
	if err != nil {
		log.Printf("error: failed to write HTTP response %s", err)
	}
}

func (ht *HTTPTransport) handleWritePayload(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
	t, err := ht.nl.Topic(ps.ByName("topic"))
	if err != nil {
		JSONErrorResponse(w, err)
		return
	}

	defer r.Body.Close()

	buf, err := ioutil.ReadAll(r.Body)
	if len(buf) == 0 || len(buf) < int(r.ContentLength) {
		JSONErrorResponse(w, netlog.ErrBadRequest)
		return
	}

	if err != nil {
		JSONErrorResponse(w, err)
		return
	}

	entry := netlog.MessageFromPayload(buf)
	buf = entry.Bytes()
	_, err = t.Write(buf)
	if err != nil {
		JSONErrorResponse(w, err)
		return
	}

	w.WriteHeader(http.StatusCreated)
}

func (ht *HTTPTransport) handleSync(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
	t, err := ht.nl.Topic(ps.ByName("topic"))
	if err != nil {
		JSONErrorResponse(w, err)
		return
	}

	err = t.Sync()
	if err != nil {
		JSONErrorResponse(w, err)
		return
	}

	JSONOKResponse(w, "synced")
}

func (ht *HTTPTransport) handleTopicInfo(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
	t, err := ht.nl.Topic(ps.ByName("topic"))
	if err != nil {
		JSONErrorResponse(w, err)
		return
	}

	blInfo, err := t.Info()
	if err != nil {
		JSONErrorResponse(w, err)
		return
	}

	blInfo.Segments = nil
	JSONResponse(w, blInfo)
}

func (ht *HTTPTransport) handleServerInfo(w http.ResponseWriter, r *http.Request, _ httprouter.Params) {
	JSONResponse(w, ht.nl.TopicList())<commit_msg>transport: display segment info<commit_after>// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at http://mozilla.org/MPL/2.0/.

package transport

import (
	"encoding/json"
	"io"
	"io/ioutil"
	"log"
	"net/http"
	"strconv"
	"time"

	"github.com/julienschmidt/httprouter"
	"golang.org/x/net/context"

	"github.com/ninibe/bigduration"
	"github.com/ninibe/netlog"
)

// NewHTTPTransport transport sets up an HTTP interface around a NetLog.
func NewHTTPTransport(nl *netlog.NetLog) *HTTPTransport {
	return &HTTPTransport{nl: nl}
}

// HTTPTransport implements an HTTP server around a NetLog.
type HTTPTransport struct {
	nl *netlog.NetLog
}

// ServeHTTP implements the http.Handler interface around a NetLog.
func (ht *HTTPTransport) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	router := httprouter.New()
	router.GET("/", ht.handleServerInfo)
	router.GET("/:topic", ht.handleTopicInfo)
	router.POST("/:topic", ht.handleCreateTopic)
	router.POST("/:topic/payload", ht.handleWritePayload)
	router.GET("/:topic/payload/:offset", ht.handleReadPayload)
	router.GET("/:topic/sync", ht.handleSync)
	router.POST("/:topic/scanner", ht.handleCreateScanner)
	//router.DELETE("/:topic/scanner", ht.handleDeleteScanner)
	router.GET("/:topic/scan", withCtx(ht.handleScanTopic))
	router.DELETE("/:topic", ht.handleDeleteTopic)
	router.ServeHTTP(w, r)
	return
}

// ctxHandle is the signature a context-friendly http handler.
type ctxHandle func(context.Context, http.ResponseWriter, *http.Request, httprouter.Params)

// withCtx is a wrapper function to inject a context into an http handler
// the context gets canceled if the http connection is closed by the client.
func withCtx(handle ctxHandle) httprouter.Handle {
	ctx := context.Background()
	return func(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
		var cancel context.CancelFunc
		ctx, cancel = context.WithCancel(ctx)
		clientGone := w.(http.CloseNotifier).CloseNotify()
		go func() {
			select {
			case <-ctx.Done():
			case <-clientGone:
				cancel()
			}
		}()

		handle(ctx, w, r, ps)
	}
}

func (ht *HTTPTransport) handleCreateTopic(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {

	var settings netlog.TopicSettings
	encoder := json.NewDecoder(r.Body)
	err := encoder.Decode(&settings)
	if err != nil && err != io.EOF {
		log.Printf(err.Error())
		JSONErrorResponse(w, netlog.ErrBadRequest)
		return
	}

	_, err = ht.nl.CreateTopic(ps.ByName("topic"), settings)
	if err != nil {
		JSONErrorResponse(w, err)
		return
	}

	w.WriteHeader(http.StatusCreated)
	JSONOKResponse(w, "topic created")
}

func (ht *HTTPTransport) handleReadPayload(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
	t, err := ht.nl.Topic(ps.ByName("topic"))
	if err != nil {
		JSONErrorResponse(w, err)
		return
	}

	defer r.Body.Close()

	offset, err := t.ParseOffset(ps.ByName("offset"))
	if err != nil {
		JSONErrorResponse(w, netlog.ErrInvalidOffset)
		return
	}

	data, err := t.Payload(offset)
	if err != nil {
		JSONErrorResponse(w, err)
		return
	}

	w.Header().Set("Content-Type", "application/octet-stream")
	_, err = w.Write(data)
	if err != nil {
		log.Printf("error: failed to write HTTP response %s", err)
	}
}

func (ht *HTTPTransport) handleWritePayload(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
	t, err := ht.nl.Topic(ps.ByName("topic"))
	if err != nil {
		JSONErrorResponse(w, err)
		return
	}

	defer r.Body.Close()

	buf, err := ioutil.ReadAll(r.Body)
	if len(buf) == 0 || len(buf) < int(r.ContentLength) {
		JSONErrorResponse(w, netlog.ErrBadRequest)
		return
	}

	if err != nil {
		JSONErrorResponse(w, err)
		return
	}

	entry := netlog.MessageFromPayload(buf)
	buf = entry.Bytes()
	_, err = t.Write(buf)
	if err != nil {
		JSONErrorResponse(w, err)
		return
	}

	w.WriteHeader(http.StatusCreated)
}

func (ht *HTTPTransport) handleSync(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
	t, err := ht.nl.Topic(ps.ByName("topic"))
	if err != nil {
		JSONErrorResponse(w, err)
		return
	}

	err = t.Sync()
	if err != nil {
		JSONErrorResponse(w, err)
		return
	}

	JSONOKResponse(w, "synced")
}

func (ht *HTTPTransport) handleTopicInfo(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
	t, err := ht.nl.Topic(ps.ByName("topic"))
	if err != nil {
		JSONErrorResponse(w, err)
		return
	}

	info, err := t.Info()
	if err != nil {
		JSONErrorResponse(w, err)
		return
	}

	JSONResponse(w, info)
}

func (ht *HTTPTransport) handleServerInfo(w http.ResponseWriter, r *http.Request, _ httprouter.Params) {
	JSONResponse(w, ht.nl.TopicList())
{'fraction_non_alphanumeric': 0.14191699394968688, 'fraction_numerical': 0.0010614584439019212, 'mean_word_length': 4.2784313725490195, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4122987', 'n_tokens_mistral': 3430, 'n_tokens_neox': 3323, 'n_words': 804}