maintner/subscribe.go

// Copyright 2017 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

package maintner

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"net/http"
	"strings"
	"time"

	"golang.org/x/build/cmd/pubsubhelper/pubsubtypes"
)

func (c *Corpus) activityChan(topic string) chan struct{} {
	c.mu.Lock()
	defer c.mu.Unlock()
	if ch, ok := c.activityChans[topic]; ok {
		return ch
	}
	if c.activityChans == nil {
		c.activityChans = map[string]chan struct{}{}
	}
	ch := make(chan struct{}) // unbuffered
	c.activityChans[topic] = ch
	return ch
}

func (c *Corpus) fire(topic string) {
	ch := c.activityChan(topic)
	select {
	case ch <- struct{}{}:
		log.Printf("Pubsub woke up sync for topic %q", topic)
	default:
		log.Printf("Pubsub event on topic %q discarded; already syncing?", topic)
	}
}

// StartPubSubHelperSubscribe starts subscribing to a
// golang.org/x/build/cmd/pubsubhelper server, such
// as https://pubsubhelper.golang.org
func (c *Corpus) StartPubSubHelperSubscribe(urlBase string) {
	go c.subscribeLoop(urlBase)
}

func (c *Corpus) subscribeLoop(urlBase string) {
	var after time.Time
	for {
		newAfter, err := c.getEvent(urlBase, after)
		if err != nil {
			log.Printf("pubsub subscribe: %v", err)
			time.Sleep(5 * time.Second)
			continue
		}
		after = newAfter
	}
}

var zt time.Time // a zero time.Time

func (c *Corpus) getEvent(urlBase string, after time.Time) (newAfter time.Time, err error) {
	var afterStr string
	if !after.IsZero() {
		afterStr = after.UTC().Format(time.RFC3339Nano)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
	defer cancel()

	req, _ := http.NewRequest("GET", urlBase+"/waitevent?after="+afterStr, nil)
	req = req.WithContext(ctx)
	res, err := http.DefaultClient.Do(req)
	if err != nil {
		return zt, err
	}
	defer res.Body.Close()
	if res.StatusCode != 200 {
		return zt, errors.New(res.Status)
	}
	var evt pubsubtypes.Event
	if err := json.NewDecoder(res.Body).Decode(&evt); err != nil {
		return zt, err
	}
	if !evt.LongPollTimeout {
		got, _ := json.MarshalIndent(evt, "", "\t")
		log.Printf("Got pubsubhelper event: %s", got)
		if gh := evt.GitHub; gh != nil {
			topic := "github:" + gh.RepoOwner + "/" + gh.Repo
			c.fire(topic)
		}
		if gr := evt.Gerrit; gr != nil {
			c.fire(gerritTopicOfEvent(gr))
		}
	}
	return evt.Time.Time(), nil
}

// Return topics like "gerrit:go.googlesource.com/build"
func gerritTopicOfEvent(gr *pubsubtypes.GerritEvent) string {
	server := gr.URL // "https://code-review.googlesource.com/11970
	if i := strings.Index(server, "//"); i != -1 {
		server = server[i+2:] // code-review.googlesource.com/11970
	}
	if i := strings.Index(server, "/"); i != -1 {
		server = server[:i] // code-review.googlesource.com
	}
	server = strings.Replace(server, "-review.googlesource.com", ".googlesource.com", 1)
	return fmt.Sprintf("gerrit:%s/%s", server, gr.Project)
}