1	/*
     2	Copyright 2011 The Perkeep Authors
     3	
     4	Licensed under the Apache License, Version 2.0 (the "License");
     5	you may not use this file except in compliance with the License.
     6	You may obtain a copy of the License at
     7	
     8	     http://www.apache.org/licenses/LICENSE-2.0
     9	
    10	Unless required by applicable law or agreed to in writing, software
    11	distributed under the License is distributed on an "AS IS" BASIS,
    12	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    13	See the License for the specific language governing permissions and
    14	limitations under the License.
    15	*/
    16	
    17	package handlers
    18	
    19	import (
    20		"context"
    21		"fmt"
    22		"io"
    23		"log"
    24		"net/http"
    25		"strconv"
    26		"time"
    27	
    28		"perkeep.org/pkg/blob"
    29		"perkeep.org/pkg/blobserver"
    30	)
    31	
    32	const defaultMaxEnumerate = 10000
    33	const defaultEnumerateSize = 100
    34	
    35	func CreateEnumerateHandler(storage blobserver.BlobEnumerator) http.Handler {
    36		return http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
    37			handleEnumerateBlobs(rw, req, storage)
    38		})
    39	}
    40	
    41	const errMsgMaxWaitSecWithAfter = "Can't use 'maxwaitsec' with 'after'.\n"
    42	
    43	func handleEnumerateBlobs(rw http.ResponseWriter, req *http.Request, storage blobserver.BlobEnumerator) {
    44		// Potential input parameters
    45		formValueLimit := req.FormValue("limit")
    46		formValueMaxWaitSec := req.FormValue("maxwaitsec")
    47		formValueAfter := req.FormValue("after")
    48	
    49		maxEnumerate := defaultMaxEnumerate
    50		if config, ok := storage.(blobserver.MaxEnumerateConfig); ok {
    51			maxEnumerate = config.MaxEnumerate()
    52		}
    53	
    54		limit := defaultEnumerateSize
    55		if formValueLimit != "" {
    56			n, err := strconv.ParseUint(formValueLimit, 10, 32)
    57			if err != nil || n > uint64(maxEnumerate) {
    58				limit = maxEnumerate
    59			} else {
    60				limit = int(n)
    61			}
    62		}
    63	
    64		waitSeconds := 0
    65		if formValueMaxWaitSec != "" {
    66			waitSeconds, _ = strconv.Atoi(formValueMaxWaitSec)
    67			if waitSeconds != 0 && formValueAfter != "" {
    68				rw.WriteHeader(http.StatusBadRequest)
    69				fmt.Fprintf(rw, errMsgMaxWaitSecWithAfter)
    70				return
    71			}
    72			switch {
    73			case waitSeconds < 0:
    74				waitSeconds = 0
    75			case waitSeconds > 30:
    76				// TODO: don't hard-code 30.  push this up into a blobserver interface
    77				// for getting the configuration of the server (ultimately a flag in
    78				// in the binary)
    79				waitSeconds = 30
    80			}
    81		}
    82	
    83		rw.Header().Set("Content-Type", "text/javascript; charset=utf-8")
    84		io.WriteString(rw, "{\n  \"blobs\": [\n")
    85	
    86		loop := true
    87		needsComma := false
    88		deadline := time.Now().Add(time.Duration(waitSeconds) * time.Second)
    89		after := ""
    90		for loop && (waitSeconds == 0 || time.Now().After(deadline)) {
    91			if waitSeconds == 0 {
    92				loop = false
    93			}
    94	
    95			blobch := make(chan blob.SizedRef, 100)
    96			resultch := make(chan error, 1)
    97			go func() {
    98				resultch <- storage.EnumerateBlobs(context.TODO(), blobch, formValueAfter, limit)
    99			}()
   100	
   101			gotBlobs := 0
   102			for sb := range blobch {
   103				gotBlobs++
   104				loop = false
   105				blobName := sb.Ref.String()
   106				if needsComma {
   107					io.WriteString(rw, ",\n")
   108				}
   109				fmt.Fprintf(rw, "    {\"blobRef\": \"%s\", \"size\": %d}",
   110					blobName, sb.Size)
   111				after = blobName
   112				needsComma = true
   113			}
   114			if gotBlobs < limit {
   115				after = ""
   116			}
   117			if err := <-resultch; err != nil {
   118				log.Printf("Error during enumerate: %v", err)
   119				fmt.Fprintf(rw, "{{{ SERVER ERROR }}}")
   120				return
   121			}
   122	
   123			if loop {
   124				blobserver.WaitForBlob(storage, deadline, nil)
   125			}
   126		}
   127		io.WriteString(rw, "\n  ]")
   128		if after != "" {
   129			fmt.Fprintf(rw, ",\n  \"continueAfter\": \"%s\"", after)
   130		}
   131		const longPollSupported = true
   132		if longPollSupported {
   133			io.WriteString(rw, ",\n  \"canLongPoll\": true")
   134		}
   135		io.WriteString(rw, "\n}\n")
   136	}