package main
import (
	"fmt"
	"encoding/json"
)
type MyData struct {
	One int
	two string
}
func main() {
	in := MyData{1,"two"}
	fmt.Printf("%#v\n",in) //prints main.MyData{One:1, two:"two"}
	encoded,_ := json.Marshal(in)
	fmt.Println(string(encoded)) //prints {"One":1}
	var out MyData
	json.Unmarshal(encoded,&out)
	fmt.Printf("%#v\n",out) //prints main.MyData{One:1, two:""}
}

type MyData struct {
	One int `json:"one"`
	Two string `json:"two"`
}

package main

import "github.com/go-martini/martini"

func main() {
    m := martini.Classic()
    m.Get("/", func() string {
    	return "Hello world!"
    })
    m.Run()
}

package main

import (
    "fmt"
    "github.com/codegangsta/inject"
)

// 自定义一个空interface{}, 必须是interface{}类型MapTo才能接受
type SpecialString interface{}

// 原定义Foo(s1,s2 string), 改成
func Foo(s1 string, s2 SpecialString) {
    fmt.Println(s1, s2.(string)) // type assertion
}

func main() {
    ij := inject.New()
    // 注入string类型的"a"
    ij.Map("a")
    // 注入SpecialString类型的"b"
    ij.MapTo("b", (*SpecialString)(nil))
    // 调用Foo函数时，会自动查找已经注入的值，并自动传递。
    // inject包的设计确保不会有两个相同类型的值（限制）
    // 若找不到会抛出运行时错误
    ij.Invoke(Foo)
}

type Str struct{
	b int
}

func main() {
	var b *Str = nil
	var a = (*Str)(nil)
    // 变量a,b都是一个类型为Str的空指针
	fmt.Print(a)
	fmt.Print(b)
}

// InterfaceOf dereferences a pointer to an Interface type.
// It panics if value is not an pointer to an interface.
func InterfaceOf(value interface{}) reflect.Type {
	t := reflect.TypeOf(value)

	// 获取实际的类型，用for循环是为了确保多级指针（指向指针的指针）
	for t.Kind() == reflect.Ptr {
		t = t.Elem()
	}

	if t.Kind() != reflect.Interface {
		panic("Called ... interface. (*MyInterface)(nil)")
	}

	return t
}

type ClassicMartini struct {
	*Martini
	Router
}

type Martini struct {
	inject.Injector
	handlers []Handler
	action   Handler
	logger   *log.Logger
}

func Classic() *ClassicMartini {
	r := NewRouter()
	m := New()
	m.Use(Logger())
	m.Use(Recovery())
	m.Use(Static("public"))
    // Mapto注入router处理器
	m.MapTo(r, (*Routes)(nil))
	m.Action(r.Handle)
	return &ClassicMartini{m, r}
}

func (c *context) handler() Handler {
	if c.index < len(c.handlers) {
		return c.handlers[c.index]
	}
    // 最后调用
	if c.index == len(c.handlers) {
		return c.action
	}
	panic("invalid index for context handler")
}

// Run the http server on a given host and port.
func (m *Martini) RunOnAddr(addr string) {
	logger := m.Injector.Get(reflect.TypeOf(m.logger)).Interface().(*log.Logger)
	logger.Printf("listening on %s (%s)\n", addr, Env)
	logger.Fatalln(http.ListenAndServe(addr, m))
}

func (m *Martini) ServeHTTP(res http.ResponseWriter, req *http.Request) {
	m.createContext(res, req).run()
}

type context struct {
	inject.Injector
	handlers []Handler
	action   Handler
	rw       ResponseWriter
	index    int
}

func (m *Martini) createContext(res http.ResponseWriter, req *http.Request) *context {
	c := &context{inject.New(), m.handlers, m.action, NewResponseWriter(res), 0}
    // 从m继承注入的变量，比如logger之类的全局变量
	c.SetParent(m)
    // 注入自己
	c.MapTo(c, (*Context)(nil))
    // 注入responce
	c.MapTo(c.rw, (*http.ResponseWriter)(nil))
    // 注入request
	c.Map(req)
	return c
}

// 最后运行Run函数，先依次运行middleware，然后运行路由总入口，参见handler()函数
func (c *context) run() {
	for c.index <= len(c.handlers) {
        // 最终会触发router.Handle
		_, err := c.Invoke(c.handler())
		if err != nil {
			panic(err)
		}
		c.index += 1

		if c.Written() {
			return
		}
	}
}

type router struct {
	// 所有的路由对象
	routes     []*route
	notFounds  []Handler
	groups     []group
	routesLock sync.RWMutex
}

type route struct {
	method   string
	regex    *regexp.Regexp
    // 所有的handle
	handlers []Handler
	pattern  string
	name     string
}

func (r *route) Handle(c Context, res http.ResponseWriter) {
	// 构建新的对象，
	context := &routeContext{c, 0, r.handlers}
    // 注入新的变量
	c.MapTo(context, (*Context)(nil))
	c.MapTo(r, (*Route)(nil))
    // 注意这里调用的是routeContext的run
	context.run()
}

type routeContext struct {
	// 继承了最初的Context，后者又继承了Martini注入的变量
	Context
	index    int
	handlers []Handler
}

// 依次触发所有的handle
func (r *routeContext) run() {
	for r.index < len(r.handlers) {
		handler := r.handlers[r.index]
        // 依次触发回调
		vals, err := r.Invoke(handler)
		if err != nil {
			panic(err)
		}
		r.index += 1

		// if the handler returned something, write it to the http response
		if len(vals) > 0 {
			ev := r.Get(reflect.TypeOf(ReturnHandler(nil)))
			handleReturn := ev.Interface().(ReturnHandler)
			handleReturn(r, vals)
		}

		if r.Written() {
			return
		}
	}
}

ws.GET("/logs/:owner/:name/:build/:number",
			session.SetRepo(),
			session.SetPerm(),
			session.MustPull,
			server.LogStream,
		)

func SetRepo() gin.HandlerFunc {
	return func(c *gin.Context) {
		var (
			owner = c.Param("owner")
			name  = c.Param("name")
			user  = User(c)
		)

// LogStream streams the build log output to the client.
func LogStream(c *gin.Context) {
	repo := session.Repo(c)
	buildn, _ := strconv.Atoi(c.Param("build"))
	jobn, _ := strconv.Atoi(c.Param("number"))

func PostBuild(c *gin.Context) {
	// 获得后端实例
	remote_ := remote.FromContext(c)

	// 从数据库中获取的resp对象，通过前置的middleware完成了查询
	repo := session.Repo(c)
	fork := c.DefaultQuery("fork", "false")

	// 获得当前用户
	user, err := store.GetUser(c, repo.UserID)
	if err != nil {
		log.Errorf("failure to find repo owner %s. %s", repo.FullName, err)
		c.AbortWithError(500, err)
		return
	}

	// 获得build number
	build, err := store.GetBuildNumber(c, repo, num)
	if err != nil {
		log.Errorf("failure to get build %d. %s", num, err)
		c.AbortWithError(404, err)
		return
	}

	// 拉取.drone.yml文件
	// fetch the .drone.yml file from the database
	config := ToConfig(c)
	raw, err := remote_.File(user, repo, build, config.Yaml)
	if err != nil {
		log.Errorf("failure to get build config for %s. %s", repo.FullName, err)
		c.AbortWithError(404, err)
		return
	}

	// 检查是否有签名文件
	// Fetch secrets file but don't exit on error as it's optional
	sec, err := remote_.File(user, repo, build, config.Shasum)
	if err != nil {
		log.Debugf("cannot find build secrets for %s. %s", repo.FullName, err)
	}

	// 验证签名
	signature, err := jose.ParseSigned(string(sec))
	if err != nil {
		log.Debugf("cannot parse .drone.yml.sig file. %s", err)
	} else if len(sec) == 0 {
		log.Debugf("cannot parse .drone.yml.sig file. empty file")
	} else {
		signed = true
		output, err := signature.Verify([]byte(repo.Hash))
		if err != nil {
			log.Debugf("cannot verify .drone.yml.sig file. %s", err)
		} else if string(output) != string(raw) {
			log.Debugf("cannot verify .drone.yml.sig file. no match. %q <> %q", string(output), string(raw))
		} else {
			verified = true
		}
	}

	// 发送请求到tast list
	client := stomp.MustFromContext(c)
	client.SendJSON("/topic/events", model.Event{
		Type:  model.Enqueued,
		Repo:  *repo,
		Build: *build,
	},
		stomp.WithHeader("repo", repo.FullName),
		stomp.WithHeader("private", strconv.FormatBool(repo.IsPrivate)),
	)

	for _, job := range jobs {
		broker, _ := stomp.FromContext(c)
		// 发送请求到agent
		broker.SendJSON("/queue/pending", &model.Work{
			Signed:    signed,
			Verified:  verified,
			User:      user,
			Repo:      repo,
			Build:     build,
			BuildLast: last,
			Job:       job,
			Netrc:     netrc,
			Yaml:      string(raw),
			Secrets:   secs,
			System:    &model.System{Link: httputil.GetURL(c.Request)},
		},
			stomp.WithHeader(
				"platform",
				yaml.ParsePlatformDefault(raw, "linux/amd64"),
			),
			stomp.WithHeaders(
				yaml.ParseLabel(raw),
			),
		)
	}
}

var signCmd = cli.Command{
	Name:  "sign",
	Usage: "creates a secure yaml file",
	Action: func(c *cli.Context) {
		if err := sign(c); err != nil {
			log.Fatalln(err)
		}
	},
	Flags: []cli.Flag{
		cli.StringFlag{
			Name:  "in",
			Usage: "input file",
			Value: ".drone.yml",
		},
		cli.StringFlag{
			Name:  "out",
			Usage: "output file signature",
			Value: ".drone.yml.sig",
		},
	},
}

func start(c *cli.Context) {
	// 初始化全局的docker实例
	docker, err := dockerclient.NewDockerClient(c.String("docker-host"), tls)
	if err != nil {
		logrus.Fatal(err)
	}

	var client *stomp.Client

	// 收到请求的回调
	handler := func(m *stomp.Message) {
		running.Add(1)
		defer func() {
			running.Done()
			client.Ack(m.Ack)
		}()

		r := pipeline{
			drone:  client,
			docker: docker,
			config: config{
				platform:   c.String("docker-os") + "/" + c.String("docker-arch"),
				timeout:    c.Duration("timeout"),
				namespace:  c.String("namespace"),
				privileged: c.StringSlice("privileged"),
				pull:       c.BoolT("pull"),
				logs:       int64(c.Int("max-log-size")) * 1000000,
				extension:  c.StringSlice("extension"),
			},
		}

		work := new(model.Work)
		m.Unmarshal(work)
		r.run(work)
	}

	for {
		// dial the drone server to establish a TCP connection.
		client, err = stomp.Dial(server)
		if err != nil {
			logger.Warningf("connection failed, retry in %v. %s", backoff, err)
			<-time.After(backoff)
			continue
		}
		opts := []stomp.MessageOption{
			stomp.WithCredentials("x-token", accessToken),
		}

		// 连接mq
		// initialize the stomp session and authenticate.
		if err = client.Connect(opts...); err != nil {
			logger.Warningf("session failed, retry in %v. %s", backoff, err)
			<-time.After(backoff)
			continue
		}

        // 订阅topic
		// subscribe to the pending build queue.
		client.Subscribe("/queue/pending", stomp.HandlerFunc(func(m *stomp.Message) {
			go handler(m) // HACK until we a channel based Subscribe implementation
		}), opts...)

// Work represents an item for work to be
// processed by a worker.
type Work struct {
	Signed    bool      `json:"signed"`
	Verified  bool      `json:"verified"`
	Yaml      string    `json:"config"`
	YamlEnc   string    `json:"secret"`
	Repo      *Repo     `json:"repo"`
	Build     *Build    `json:"build"`
	BuildLast *Build    `json:"build_last"`
	Job       *Job      `json:"job"`
	Netrc     *Netrc    `json:"netrc"`
	Keys      *Key      `json:"keys"`
	System    *System   `json:"system"`
	Secrets   []*Secret `json:"secrets"`
	User      *User     `json:"user"`
}

// Engine defines the container runtime engine.
type Engine interface {
	ContainerStart(*yaml.Container) (string, error)
	ContainerStop(string) error
	ContainerRemove(string) error
	ContainerWait(string) (*State, error)
	ContainerLogs(string) (io.ReadCloser, error)
}

ws.GET("/feed", server.EventStream)
ws.GET("/logs/:owner/:name/:build/:number",
    session.SetRepo(),
    session.SetPerm(),
    session.MustPull,
    server.LogStream,
)

func (r *pipeline) run(w *model.Work) {
	// 创建cancel channel
	cancel := make(chan bool, 1)
    // 创建docker实例
	engine := docker.NewClient(r.docker)
	// 创建agent实例
	a := agent.Agent{
    	// 发送build日志
		Update:    agent.NewClientUpdater(r.drone),
		Logger:    agent.NewClientLogger(r.drone, w.Job.ID, r.config.logs),
		Engine:    engine,
		Timeout:   r.config.timeout,
		Platform:  r.config.platform,
		Namespace: r.config.namespace,
		Escalate:  r.config.privileged,
		Extension: r.config.extension,
		Pull:      r.config.pull,
	}


	// 支持取消编译事件
	// signal for canceling the build.
	sub, err := r.drone.Subscribe("/topic/cancel", stomp.HandlerFunc(cancelFunc))
	if err != nil {
		logrus.Errorf("Error subscribing to /topic/cancel. %s", err)
	}

	a.Run(w, cancel)

}

func (a *Agent) Run(payload *model.Work, cancel <-chan bool) error {

	// 预处理
	spec, err := a.prep(payload)

	// Update是一个value为函数的回调值，用于更新任务状态，见agent.NewClientUpdater
	a.Update(payload)

    // 执行
	err = a.exec(spec, payload, cancel)

	a.Update(payload)
	return err
}

// Config represents the build configuration Yaml document.
type  struct {
	Image     string
	Build     *Build
	Workspace *Workspace
	Pipeline  []*Container
	Services  []*Container
	Volumes   []*Volume
	Networks  []*Network
}

func (a *Agent) prep(w *model.Work) (*yaml.Config, error) {

	envs := toEnv(w)
	envSecrets := map[string]string{}

	// list of secrets to interpolate in the yaml
	for _, secret := range w.Secrets {
		if (w.Verified || secret.SkipVerify) && secret.MatchEvent(w.Build.Event) {
			envSecrets[secret.Name] = secret.Value
		}
	}

	var err error
	w.Yaml, err = envsubst.Eval(w.Yaml, func(s string) string {
		env, ok := envSecrets[s]
		if !ok {
			env, _ = envs[s]
		}
		if strings.Contains(env, "\n") {
			env = fmt.Sprintf("%q", env)
		}
		return env
	})
	if err != nil {
		return nil, err
	}


	conf, err := yaml.ParseString(w.Yaml)
	if err != nil {
		return nil, err
	}

	return conf, nil
}

type  struct {
	Pipeline  []*Container
	Services  []*Container
}

// Container defines a Docker container.
type Container struct {
	ID             string
	Name           string
	Image          string
	Build          string
	Pull           bool
	AuthConfig     Auth
	Detached       bool
	Disabled       bool
	Privileged     bool
	WorkingDir     string
	Environment    map[string]string
	Labels         map[string]string
	Entrypoint     []string
	Command        []string
	Commands       []string
	ExtraHosts     []string
	Volumes        []string
	VolumesFrom    []string
	Devices        []string
	Network        string
	DNS            []string
	DNSSearch      []string
	MemSwapLimit   int64
	MemLimit       int64
	ShmSize        int64
	CPUQuota       int64
	CPUShares      int64
	CPUSet         string
	OomKillDisable bool
	Constraints    Constraints

	Vargs map[string]interface{}
}

// Clone transforms the Yaml to include a clone step.
func Clone(c *yaml.Config, plugin string) error {
	switch plugin {
	case "", "git":
		plugin = "plugins/git:latest"
	case "hg":
		plugin = "plugins/hg:latest"
	}

	for _, p := range c.Pipeline {
		if p.Name == clone {
			if p.Image == "" {
				p.Image = plugin
			}
			return nil
		}
	}

	s := &yaml.Container{
		Image: plugin,
		Name:  clone,
	}

	c.Pipeline = append([]*yaml.Container{s}, c.Pipeline...)
	return nil
}

// lookup ambassador configuration by architecture and os
var lookupAmbassador = map[string]ambassador{
	"linux/amd64": {
		image:      "busybox:latest",
		entrypoint: []string{"/bin/sleep"},
		command:    []string{"86400"},
	},
	"linux/arm": {
		image:      "armhf/alpine:latest",
		entrypoint: []string{"/bin/sleep"},
		command:    []string{"86400"},
	},
}

// Pod transforms the containers in the Yaml to use Pod networking, where every
// container shares the localhost connection.
func Pod(c *yaml.Config, platform string) error {

	rand := base64.RawURLEncoding.EncodeToString(
		securecookie.GenerateRandomKey(8),
	)

	// choose the ambassador configuration based on os and architecture
	conf, ok := lookupAmbassador[platform]
	if !ok {
		conf = defaultAmbassador
	}

    for _, container := range containers {
		container.VolumesFrom = append(container.VolumesFrom, ambassador.ID)
		if container.Network == "" {
			container.Network = network
		}
	}

// Pipeline creates a build Pipeline using the specific configuration for
// the given Yaml specification.
func (c *Config) Pipeline(spec *yaml.Config) *Pipeline {

	pipeline := Pipeline{
		engine: c.Engine,
		pipe:   make(chan *Line, c.Buffer),
		next:   make(chan error),
		done:   make(chan error),
	}

	var containers []*yaml.Container
	containers = append(containers, spec.Services...)
	containers = append(containers, spec.Pipeline...)

	for _, c := range containers {
		if c.Disabled {
			continue
		}
		next := &element{Container: c}
		if pipeline.head == nil {
			pipeline.head = next
			pipeline.tail = next
		} else {
			pipeline.tail.next = next
			pipeline.tail = next
		}
	}

	go func() {
		pipeline.next <- nil
	}()

	return &pipeline
}

func (a *Agent) exec(spec *yaml.Config, payload *model.Work, cancel <-chan bool) error {

	conf := build.Config{
		Engine: a.Engine,
		Buffer: 500,
	}

	pipeline := conf.Pipeline(spec)
	defer pipeline.Teardown()

	// setup the build environment
	if err := pipeline.Setup(); err != nil {
		return err
	}

	replacer := NewSecretReplacer(payload.Secrets)
	timeout := time.After(time.Duration(payload.Repo.Timeout) * time.Minute)

	for {
		select {
		case <-pipeline.Done():
			return pipeline.Err()
		case <-cancel:
			pipeline.Stop()
			return fmt.Errorf("termination request received, build cancelled")
		case <-timeout:
			pipeline.Stop()
			return fmt.Errorf("maximum time limit exceeded, build cancelled")
		case <-time.After(a.Timeout):
			pipeline.Stop()
			return fmt.Errorf("terminal inactive for %v, build cancelled", a.Timeout)
		case <-pipeline.Next():

			// TODO(bradrydzewski) this entire block of code should probably get
			// encapsulated in the pipeline.
			status := model.StatusSuccess
			if pipeline.Err() != nil {
				status = model.StatusFailure
			}
			// updates the build status passed into each container. I realize this is
			// a bit out of place and will work to resolve.
			pipeline.Head().Environment["DRONE_BUILD_STATUS"] = status

			if !pipeline.Head().Constraints.Match(
				a.Platform,
				payload.Build.Deploy,
				payload.Build.Event,
				payload.Build.Branch,
				status, payload.Job.Environment) { // TODO: fix this whole section

				pipeline.Skip()
			} else {
				pipeline.Exec()
			}
		case line := <-pipeline.Pipe():
			line.Out = replacer.Replace(line.Out)
			a.Logger(line)
		}
	}
}

func main() {
	envflag.Parse()

	app := cli.NewApp()
	app.Name = "drone"
	app.Version = version.Version
	app.Usage = "command line utility"
	app.Flags = []cli.Flag{
		cli.StringFlag{
			Name:   "t, token",
			Usage:  "server auth token",
			EnvVar: "DRONE_TOKEN",
		},
		cli.StringFlag{
			Name:   "s, server",
			Usage:  "server location",
			EnvVar: "DRONE_SERVER",
		},
	}
	app.Commands = []cli.Command{
		agent.AgentCmd,
		agentsCmd,
		buildCmd,
		deployCmd,
		execCmd,
		infoCmd,  //-------------------
		secretCmd,
		serverCmd,
		signCmd,
		repoCmd,
		userCmd,
		orgCmd,
		globalCmd,
	}

	app.Run(os.Args)
}

var infoCmd = cli.Command{
	Name:  "info",
	Usage: "show information about the current user",
	Action: func(c *cli.Context) {
		if err := info(c); err != nil {
			log.Fatalln(err)
		}
	},
	Flags: []cli.Flag{
		cli.StringFlag{
			Name:  "format",
			Usage: "format output",
			Value: tmplUserInfo,
		},
	},
}

func userList(c *cli.Context) error {
	client, err := newClient(c)
	if err != nil {
		return err
	}

	users, err := client.UserList() //---------------------
	if err != nil || len(users) == 0 {
		return err
	}

// Client is used to communicate with a Drone server.
type Client interface {
	// Self returns the currently authenticated user.
	Self() (*model.User, error)

	// User returns a user by login.
	User(string) (*model.User, error)

	// UserList returns a list of all registered users.
	UserList() ([]*model.User, error)

const (
	pathUsers         = "%s/api/users"
)

// UserList returns a list of all registered users.
func (c *client) UserList() ([]*model.User, error) {
	var out []*model.User
	uri := fmt.Sprintf(pathUsers, c.base)

	log.Print("user list uri %s\n",uri)
	err := c.get(uri, &out)
	return out, err
}

func server(c *cli.Context) error {

	// setup the server and start the listener
	handler := router.Load(
		ginrus.Ginrus(logrus.StandardLogger(), time.RFC3339, true),
		middleware.Version,
		middleware.Config(c),
		middleware.Cache(c),
		middleware.Store(c),
		middleware.Remote(c),
		middleware.Agents(c),
		middleware.Broker(c),
	)

// Load loads the router
func Load(middleware ...gin.HandlerFunc) http.Handler {
	e.Use(header.NoCache)
	e.Use(header.Options)
	e.Use(header.Secure)
	e.Use(middleware...)
	e.Use(session.SetUser())
	e.Use(token.Refresh)

	e.GET("/login", server.ShowLogin)
	e.GET("/login/form", server.ShowLoginForm)
	e.GET("/logout", server.GetLogout)
	e.NoRoute(server.ShowIndex)

	// TODO above will Go away with React UI

	user := e.Group("/api/user")
	{
		user.Use(session.MustUser())
		user.GET("", server.GetSelf)
		user.GET("/feed", server.GetFeed)
		user.GET("/repos", server.GetRepos)
		user.GET("/repos/remote", server.GetRemoteRepos)
		user.POST("/token", server.PostToken)
		user.DELETE("/token", server.DeleteToken)
	}

repos := e.Group("/api/repos/:owner/:name")
{
	repos.POST("", server.PostRepo)

	repo := repos.Group("")
	{
		repo.Use(session.SetRepo())
		repo.Use(session.SetPerm())
		repo.Use(session.MustPull)

		repo.GET("", server.GetRepo)
	}
}

func SetUser() gin.HandlerFunc {
	return func(c *gin.Context) {
		var user *model.User
		// 获取token，并验证
		t, err := token.ParseRequest(c.Request, func(t *token.Token) (string, error) {
			var err error
			// 继续查询数据库是否存在用户
			user, err = store.GetUserLogin(c, t.Text)
			return user.Hash, err
		})
		if err == nil {
			confv := c.MustGet("config")
			if conf, ok := confv.(*model.Config); ok {
				// 是否admin
				user.Admin = conf.IsAdmin(user)
			}

			// 最后设置当前的user到gin.Context
			c.Set("user", user)

			// 普通的web token，需要验证csrf token
			if t.Kind == token.SessToken {
				err = token.CheckCsrf(c.Request, func(t *token.Token) (string, error) {
					return user.Hash, nil
				})
			}
		}
		c.Next()
	}
}

func User(c *gin.Context) *model.User {
	// 从gin.Context获取当前的user，可能不存在
	// 参考对比 SetUser c.Set("user", user)
	v, ok := c.Get("user")
	if !ok {
		return nil
	}
	u, ok := v.(*model.User)
	if !ok {
		return nil
	}
	return u
}

func MustRepoAdmin() gin.HandlerFunc {
	return func(c *gin.Context) {
		user := User(c)
		perm := Perm(c)
		switch {
		case user == nil:
			c.String(401, "User not authorized")
			c.Abort()
		case perm.Admin == false:
			c.String(403, "User not authorized")
			c.Abort()
		default:
			c.Next()
		}
	}
}

func MustUser() gin.HandlerFunc {
	return func(c *gin.Context) {
		user := User(c)
		switch {
		case user == nil:
			c.String(401, "User not authorized")
			c.Abort()
		default:
			c.Next()
		}
	}
}

func Config(cli *cli.Context) gin.HandlerFunc {
	v := setupConfig(cli)
	return func(c *gin.Context) {
		c.Set(configKey, v)
	}
}

// helper function to create the configuration from the CLI context.
func setupConfig(c *cli.Context) *model.Config {
	return &model.Config{
		Open:   c.Bool("open"),
		Yaml:   c.String("yaml"),
		Shasum: c.String("yaml") + ".sig",
		Secret: c.String("agent-secret"),
		Admins: sliceToMap(c.StringSlice("admin")),
		Orgs:   sliceToMap(c.StringSlice("orgs")),
	}
}

auth := e.Group("/authorize")
{
    auth.GET("", server.GetLogin)
    auth.POST("", server.GetLogin)
    auth.POST("/token", server.GetLoginToken)
}

func GetLogin(c *gin.Context) {
    // 获取全局配置对象
	config := ToConfig(c)

	// get the user from the database
	u, err := store.GetUserLogin(c, tmpuser.Login)
	if err != nil {

		// if self-registration is disabled we should return a not authorized error
        // 检查
		if !config.Open && !config.IsAdmin(tmpuser) {
			logrus.Errorf("cannot register %s. registration closed", tmpuser.Login)
			c.Redirect(303, "/login?error=access_denied")
			return
		}
	}
}

// Store is a middleware function that initializes the Datastore and attaches to
// the context of every http.Request.
func Store(cli *cli.Context) gin.HandlerFunc {
	v := setupStore(cli)
	return func(c *gin.Context) {
    	// 将数据库对象设置到gin.Context
		store.ToContext(c, v)
		c.Next()
	}
}

// helper function to create the datastore from the CLI context.
func setupStore(c *cli.Context) store.Store {
	return datastore.New(
		c.String("driver"),
		c.String("datasource"),
	)
}

// New creates a database connection for the given driver and datasource
// and returns a new Store.
func New(driver, config string) store.Store {
	return From(
		open(driver, config),
	)
}

// helper function to setup the meddler default driver
// based on the selected driver name.
func setupMeddler(driver string) {
	// 根据配置，获取实际的数据库对象
	switch driver {
	case "sqlite3":
		meddler.Default = meddler.SQLite
	case "mysql":
		meddler.Default = meddler.MySQL
	case "postgres":
		meddler.Default = meddler.PostgreSQL
	}
}

// helper function to setup the databsae by performing
// automated database migration steps.
func setupDatabase(driver string, db *sql.DB) error {
	var migrations = &migrate.AssetMigrationSource{
		Asset:    ddl.Asset,
		AssetDir: ddl.AssetDir,
		Dir:      driver,
	}
    // UP 确保最新，如果没有就创建表什么的
	_, err := migrate.Exec(db, driver, migrations, migrate.Up)
	return err
}

// open opens a new database connection with the specified
// driver and connection string and returns a store.
func open(driver, config string) *sql.DB {
	db, err := sql.Open(driver, config)
	if err != nil {
		logrus.Errorln(err)
		logrus.Fatalln("database connection failed")
	}
	if driver == "mysql" {
		// per issue https://github.com/go-sql-driver/mysql/issues/257
		db.SetMaxIdleConns(0)
	}

    // 设置数据库
	setupMeddler(driver)

    // 确保数据库/表是最新的
	if err := setupDatabase(driver, db); err != nil {
		logrus.Errorln(err)
		logrus.Fatalln("migration failed")
	}
}

func (db *datastore) GetUserLogin(login string) (*model.User, error) {
	var usr = new(model.User)
	var err = meddler.QueryRow(db, usr, rebind(userLoginQuery), login)
	return usr, err
}

const userLoginQuery = `
SELECT *
FROM users
WHERE user_login=?
LIMIT 1
`

type Store interface {
	// GetUser gets a user by unique ID.
	GetUser(int64) (*model.User, error)

	// GetUserLogin gets a user by unique Login name.
	GetUserLogin(string) (*model.User, error)

	// GetUserList gets a list of all users in the system.
	GetUserList() ([]*model.User, error)

type Remote interface {
	// Login authenticates the session and returns the
	// remote user details.
	Login(w http.ResponseWriter, r *http.Request) (*model.User, error)

	// Auth authenticates the session and returns the remote user
	// login for the given token and secret
	Auth(token, secret string) (string, error)

// Remote is a middleware function that initializes the Remote and attaches to
// the context of every http.Request.
func Remote(c *cli.Context) gin.HandlerFunc {
	v, err := setupRemote(c)
	if err != nil {
		logrus.Fatalln(err)
	}
	return func(c *gin.Context) {
		remote.ToContext(c, v)
	}
}

// helper function to setup the remote from the CLI arguments.
func setupRemote(c *cli.Context) (remote.Remote, error) {
	switch {
	case c.Bool("github"):
		return setupGithub(c)
	case c.Bool("gitlab"):
		return setupGitlab(c)
	case c.Bool("bitbucket"):
		return setupBitbucket(c)
	case c.Bool("stash"):
		return setupStash(c)
	case c.Bool("gogs"):
		return setupGogs(c)
	default:
		return nil, fmt.Errorf("version control system not configured")
	}
}

ws := e.Group("/ws")
{
    ws.GET("/broker", server.Broker)
    ws.GET("/feed", server.EventStream)
    ws.GET("/logs/:owner/:name/:build/:number",
        session.SetRepo(),
        session.SetPerm(),
        session.MustPull,
        server.LogStream,
    )
}

// LogStream streams the build log output to the client.
func LogStream(c *gin.Context) {
	// 创建server
	ws, err := upgrader.Upgrade(c.Writer, c.Request, nil)
	if err != nil {
		if _, ok := err.(websocket.HandshakeError); !ok {
			logrus.Errorf("Cannot upgrade websocket. %s", err)
		}
		return
	}

    // 创建沟通的channel
	logs := make(chan []byte)
	done := make(chan bool)
	var eof bool
	dest := fmt.Sprintf("/topic/logs.%d", job.ID)
	client, _ := stomp.FromContext(c)

    // 订阅特定的topic，用来接收agent通过drone/mq发送过来的日志
	sub, err := client.Subscribe(dest, stomp.HandlerFunc(func(m *stomp.Message) {
		if m.Header.GetBool("eof") {
			eof = true
			done <- true
		} else if eof {
			return
		} else {
            // 在mq的回调中 通过golang的channel传递出来
			logs <- m.Body
		}
		m.Release()
	}))

	for {
		select {
		case buf := <-logs:
            // 收到消息后，通过ws发送到浏览器客户端
			ws.SetWriteDeadline(time.Now().Add(writeWait))
			ws.WriteMessage(websocket.TextMessage, buf)
		case <-done:
			return
		case <-ticker.C:
			err := ws.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(writeWait))
			if err != nil {
				return
			}
		}
	}
}

// Broker is a middleware function that initializes the broker
// and adds the broker client to the request context.
func Broker(cli *cli.Context) gin.HandlerFunc {
    // 参见环境变量DRONE_SECRET
	secret := cli.String("agent-secret")
	if secret == "" {
		logrus.Fatalf("fatal error. please provide the DRONE_SECRET")
	}

    // 启动mq server
	broker := server.NewServer(
		server.WithCredentials("x-token", secret),
	)
    // 创建mq client
	client := broker.Client()

	var once sync.Once
	return func(c *gin.Context) {
        // 设置到gin.Context
		c.Set(serverKey, broker)
		c.Set(clientKey, client)
		once.Do(func() {
			// this is some really hacky stuff
			// turns out I need to do some refactoring
			// don't judge!
			// will fix in 0.6 release
			ctx := c.Copy()
			client.Connect(
				stomp.WithCredentials("x-token", secret),
			)
			client.Subscribe("/queue/updates", stomp.HandlerFunc(func(m *stomp.Message) {
				go handlers.HandleUpdate(ctx, m.Copy())
			}))
		})
	}
}

func start(c *cli.Context) {
	var client *stomp.Client
	for {
		// dial the drone server to establish a TCP connection.
		client, err = stomp.Dial(server)
		if err != nil {
			logger.Warningf("connection failed, retry in %v. %s", backoff, err)
			<-time.After(backoff)
			continue
		}
		opts := []stomp.MessageOption{
			stomp.WithCredentials("x-token", accessToken),
		}

		// initialize the stomp session and authenticate.
		if err = client.Connect(opts...); err != nil {
			logger.Warningf("session failed, retry in %v. %s", backoff, err)
			<-time.After(backoff)
			continue
		}

// Broker handles connections to the embedded message broker.
func Broker(c *gin.Context) {
	broker := c.MustGet("broker").(http.Handler)
	broker.ServeHTTP(c.Writer, c.Request)
}

func (r *pipeline) run(w *model.Work) {
	a := agent.Agent{
		Update:    agent.NewClientUpdater(r.drone),
		Logger:    agent.NewClientLogger(r.drone, w.Job.ID, r.config.logs),
		Engine:    engine,
		Timeout:   r.config.timeout,
		Platform:  r.config.platform,
		Namespace: r.config.namespace,
		Escalate:  r.config.privileged,
		Extension: r.config.extension,
		Pull:      r.config.pull,
	}

// NewClientUpdater returns an updater that sends updated build details
// to the drone server.
func NewClientUpdater(client *stomp.Client) UpdateFunc {
	return func(w *model.Work) {
		err := client.SendJSON("/queue/updates", w)
		if err != nil {
			logger.Warningf("Error updating %s/%s#%d.%d. %s",
				w.Repo.Owner, w.Repo.Name, w.Build.Number, w.Job.Number, err)
		}
		if w.Job.Status != model.StatusRunning {
			var dest = fmt.Sprintf("/topic/logs.%d", w.Job.ID)
			var opts = []stomp.MessageOption{
				stomp.WithHeader("eof", "true"),
				stomp.WithRetain("all"),
			}

			if err := client.Send(dest, []byte("eof"), opts...); err != nil {
				logger.Warningf("Error sending eof %s/%s#%d.%d. %s",
					w.Repo.Owner, w.Repo.Name, w.Build.Number, w.Job.Number, err)
			}
		}
	}
}

func AssetFS() *assetfs.AssetFS {
	for k := range _bintree.Children {
		return &assetfs.AssetFS{
			Asset: Asset, 
			AssetDir: AssetDir, 
			AssetInfo: AssetInfo, Prefix: k}
	}
	panic("unreachable")
}

fs := http.FileServer(dist.AssetFS())
e.GET("/static/*filepath", func(c *gin.Context) {
    fs.ServeHTTP(c.Writer, c.Request)
})

package template

//go:generate go-bindata -pkg template -o template_gen.go files/

// Code generated by go-bindata.
// sources:
// files/index.html
// files/login.html
// files/logout.html
// DO NOT EDIT!

package template

e.SetHTMLTemplate(template.Load())

package ddl
//go:generate go-bindata -pkg ddl -o ddl_gen.go sqlite3/ mysql/ postgres/

// helper function to setup the remote from the CLI arguments.
func setupRemote(c *cli.Context) (remote.Remote, error) {
	switch {
	case c.Bool("github"):
		return setupGithub(c)
	case c.Bool("gitlab"):
		return setupGitlab(c)
	case c.Bool("bitbucket"):
		return setupBitbucket(c)
	case c.Bool("stash"):
		return setupStash(c)
	case c.Bool("gogs"):
		return setupGogs(c)
	default:
		return nil, fmt.Errorf("version control system not configured")
	}
}

// helper function to setup the meddler default driver
// based on the selected driver name.
func setupMeddler(driver string) {
	switch driver {
	case "sqlite3":
		meddler.Default = meddler.SQLite
	case "mysql":
		meddler.Default = meddler.MySQL
	case "postgres":
		meddler.Default = meddler.PostgreSQL
	}
}

// if this is a session token (ie not the API token)
// this means the user is accessing with a web browser,
// so we should implement CSRF protection measures.
if t.Kind == token.SessToken {
    err = token.CheckCsrf(c.Request, func(t *token.Token) (string, error) {
        return user.Hash, nil
    })
    // if csrf token validation fails, exit immediately
    // with a not authorized error.
    if err != nil {
        c.AbortWithStatus(http.StatusUnauthorized)
        return
    }
}

const (
	UserToken  = "user"
	SessToken  = "sess"
	HookToken  = "hook"
	CsrfToken  = "csrf"
	AgentToken = "agent"
)

user := e.Group("/api/user")
{
    user.POST("/token", server.PostToken)
    user.DELETE("/token", server.DeleteToken)
}

// src/github.com/drone/drone/router/router.go
e.GET("/login", server.ShowLogin)
e.GET("/login/form", server.ShowLoginForm)
e.GET("/logout", server.GetLogout)

// src/github.com/drone/drone/server/pages.go

// ShowLoginForm displays a login form for systems like Gogs that do not
// yet support oauth workflows.
func ShowLoginForm(c *gin.Context) {
	c.HTML(200, "login.html", gin.H{})
}

// src/github.com/drone/drone/server/template/files/login.html