Backend Query

// Copyright (C) 2016 Sebastian 'tokkee' Harl // All rights reserved. // // Redistribution and use in source and binary forms, with or without // modification, are permitted provided that the following conditions // are met: // 1. Redistributions of source code must retain the above copyright // notice, this list of conditions and the following disclaimer. // 2. Redistributions in binary form must reproduce the above copyright // notice, this list of conditions and the following disclaimer in the // documentation and/or other materials provided with the distribution. // // THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS // ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED // TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR // PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDERS OR // CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, // EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, // PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; // OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, // WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR // OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF // ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. // ui is a web-based UI interacting with the backend server. package main import ( "bytes" "errors" "flag" "fmt" "html/template" "io" "log" "net/http" "strings" "sync" "time" "golang.org/x/net/context" "google.golang.org/grpc" pb "tokkee.org/go-talk/grpc/proto/backend" ) var ( listen = flag.String("listen", ":8080", "listen address") backend = flag.String("backend", "localhost:50051", "backend server address") queryTmpl = template.Must(template.New("query").Parse(` Backend Query {{range .Responses}}

{{.Request}} => {{.Type}}: {{.N}}

{{end}} `)) ) type server struct { c pb.BackendClient } type queryResponse struct { Request string Type string N int64 } // runQueries executes multiple queries, separated by semicolon, in parallel. func (s *server) runQueries(ctx context.Context, query string) ([]queryResponse, error) { requests := strings.Split(query, ";") for i, r := range requests { requests[i] = strings.TrimSpace(r) } responses := make([]queryResponse, len(requests)) errCh := make(chan error, len(requests)) for i, req := range requests { go func(i int, req string) { res, err := s.c.Query(ctx, &pb.QueryRequest{Query: req}) defer func() { errCh <- err }() if err != nil { return } responses[i] = queryResponse{ Request: req, Type: res.Type, N: res.N, } }(i, req) } timeout := time.After(50 * time.Millisecond) for _ = range requests { select { case err := <-errCh: if err != nil { return nil, err } case <-timeout: return nil, errors.New("request timed out") } } return responses, nil } func (s *server) query(ctx context.Context, w http.ResponseWriter, r *http.Request) { data := &struct { Query string Responses []queryResponse }{} data.Query = r.Form.Get("q") if data.Query != "" { if r.Method != "POST" { // RFC 2616 requires us to set the "Allow" header. w.Header().Add("Allow", "POST") http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return } // TODO: use and validate XSRF tokens res, err := s.runQueries(ctx, data.Query) if err != nil { http.Error(w, fmt.Sprintf("query failed: %v", err), http.StatusBadRequest) return } data.Responses = res } var buf bytes.Buffer if err := queryTmpl.Execute(&buf, data); err != nil { http.Error(w, fmt.Sprintf("template error: %v", err), http.StatusInternalServerError) return } io.Copy(w, &buf) } type handler func(context.Context, http.ResponseWriter, *http.Request) // A mux multiplexes incoming requests based on the first part of the incoming // request URI. It differs from http.ServerMux in that it only support routing // based on the first part of the request path and it adds context handling // and form parsing. mux implements http.Handler. type mux struct { mu sync.RWMutex handlers map[string]handler } // Handle registers a handler for the specified path. It panics if a handler // has already been registered for the same path or if the path includes a // slash. func (m *mux) Handle(path string, h handler) { if strings.Index(path, "/") != -1 { panic("invalid path: " + path) } if h == nil { panic("invalid nil handler") } m.mu.Lock() defer m.mu.Unlock() if m.handlers == nil { m.handlers = make(map[string]handler) } if m.handlers[path] != nil { panic(fmt.Sprintf("duplicate handlers registered for %q", path)) } m.handlers[path] = h } // ServeHTTP handles incoming requests using the registered handlers and takes // care of request-specific setup (context management and form parsing). func (m *mux) ServeHTTP(w http.ResponseWriter, r *http.Request) { path := r.URL.Path if len(path) > 0 && path[0] == '/' { path = path[1:] } if f := strings.Split(path, "/"); len(f) > 1 { path = f[0] } if err := r.ParseForm(); err != nil { log.Printf("Invalid URL query: %v", err) http.Error(w, fmt.Sprintf("invalid URL query: %v", err), http.StatusBadRequest) return } m.mu.RLock() h := m.handlers[path] m.mu.RUnlock() if h == nil { log.Printf("Not found: %q (%v)", path, r.URL) http.NotFound(w, r) return } h(context.Background(), w, r) } func main() { flag.Parse() conn, err := grpc.Dial(*backend, grpc.WithInsecure()) if err != nil { log.Fatalf("Failed to connect to backend server at %s: %v", *backend, err) } defer conn.Close() s := &server{c: pb.NewBackendClient(conn)} m := &mux{} m.Handle("query", s.query) log.Printf("Listening on %s ...", *listen) log.Fatal(http.ListenAndServe(*listen, m)) }