• 设为首页
  • 点击收藏
  • 手机版
    手机扫一扫访问
    迪恩网络手机版
  • 关注官方公众号
    微信扫一扫关注
    迪恩网络公众号
登陆 注册
描述文字

Golang httprouter.Params类代码示例

原作者: [db:作者] 来自: [db:来源] 收藏 邀请

本文整理汇总了Golang中github.com/funkygao/httprouter.Params类的典型用法代码示例。如果您正苦于以下问题：Golang Params类的具体用法？Golang Params怎么用？Golang Params使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。



在下文中一共展示了Params类的18个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Golang代码示例。

示例1: pubRawHandler

//go:generate goannotation $GOFILE
// @rest GET /v1/raw/msgs/:topic/:ver
// tells client how to pub in raw mode: how to connect directly to kafka
func (this *manServer) pubRawHandler(w http.ResponseWriter, r *http.Request, params httprouter.Params) {
	ver := params.ByName(UrlParamVersion)
	topic := params.ByName(UrlParamTopic)
	appid := r.Header.Get(HttpHeaderAppid)
	realIp := getHttpRemoteIp(r)

	if err := manager.Default.OwnTopic(appid, r.Header.Get(HttpHeaderPubkey), topic); err != nil {
		log.Error("pub raw[%s] %s(%s) {topic:%s, ver:%s}: %s",
			appid, r.RemoteAddr, realIp, topic, ver, err)

		writeAuthFailure(w, err)
		return
	}

	cluster, found := manager.Default.LookupCluster(appid)
	if !found {
		writeBadRequest(w, "invalid appid")
		return
	}

	log.Info("pub raw[%s] %s(%s) {topic:%s ver:%s}", appid, r.RemoteAddr, realIp, topic, ver)

	var out = map[string]string{
		"store":   store.DefaultPubStore.Name(),
		"topic":   manager.Default.KafkaTopic(appid, topic, ver),
		"brokers": strings.Join(meta.Default.ZkCluster(cluster).OnlyNamedBrokerList(), ","),
	}
	b, _ := json.Marshal(out)
	w.Write(b)
}
开发者ID:funkygao，项目名称:gafka，代码行数:33，代码来源:handler_man_pub.go


示例2: deleteJobHandler

// DELETE /v1/jobs/:topic/:ver?id=22323
func (this *pubServer) deleteJobHandler(w http.ResponseWriter, r *http.Request, params httprouter.Params) {
	appid := r.Header.Get(HttpHeaderAppid)
	topic := params.ByName(UrlParamTopic)
	ver := params.ByName(UrlParamVersion)
	realIp := getHttpRemoteIp(r)
	if err := manager.Default.OwnTopic(appid, r.Header.Get(HttpHeaderPubkey), topic); err != nil {
		log.Error("-job[%s] %s(%s) {topic:%s, ver:%s} %s",
			appid, r.RemoteAddr, realIp, topic, ver, err)

		writeAuthFailure(w, err)
		return
	}

	_, found := manager.Default.LookupCluster(appid)
	if !found {
		log.Error("-job[%s] %s(%s) {topic:%s, ver:%s} cluster not found",
			appid, r.RemoteAddr, realIp, topic, ver)

		writeBadRequest(w, "invalid appid")
		return
	}

	jobId := r.URL.Query().Get("id")
	if len(jobId) < 18 { // jobId e,g. 341647700585877504
		writeBadRequest(w, "invalid job id")
		return
	}

	if err := job.Default.Delete(appid, manager.Default.KafkaTopic(appid, topic, ver), jobId); err != nil {
		if err == job.ErrNothingDeleted {
			// race failed, actor worker wins
			log.Warn("-job[%s] %s(%s) {topic:%s, ver:%s jid:%s} %v",
				appid, r.RemoteAddr, realIp, topic, ver, jobId, err)

			w.WriteHeader(http.StatusConflict)
			w.Write([]byte{})
			return
		}

		log.Error("-job[%s] %s(%s) {topic:%s, ver:%s jid:%s} %v",
			appid, r.RemoteAddr, realIp, topic, ver, jobId, err)

		writeServerError(w, err.Error())
		return
	}

	if Options.AuditPub {
		this.auditor.Trace("-job[%s] %s(%s) {topic:%s ver:%s UA:%s jid:%s}",
			appid, r.RemoteAddr, realIp, topic, ver, r.Header.Get("User-Agent"), jobId)
	}

	w.Write(ResponseOk)
}
开发者ID:funkygao，项目名称:gafka，代码行数:54，代码来源:handler_job.go


示例3: createJobHandler

// @rest POST /v1/jobs/:appid/:topic/:ver
func (this *manServer) createJobHandler(w http.ResponseWriter, r *http.Request, params httprouter.Params) {
	topic := params.ByName(UrlParamTopic)
	if !manager.Default.ValidateTopicName(topic) {
		log.Warn("illegal topic: %s", topic)

		writeBadRequest(w, "illegal topic")
		return
	}

	realIp := getHttpRemoteIp(r)

	if !this.throttleAddTopic.Pour(realIp, 1) {
		writeQuotaExceeded(w)
		return
	}

	hisAppid := params.ByName(UrlParamAppid)
	appid := r.Header.Get(HttpHeaderAppid)
	pubkey := r.Header.Get(HttpHeaderPubkey)
	ver := params.ByName(UrlParamVersion)
	if !manager.Default.AuthAdmin(appid, pubkey) {
		log.Warn("suspicous create job %s(%s) {appid:%s pubkey:%s topic:%s ver:%s}",
			r.RemoteAddr, realIp, appid, pubkey, topic, ver)

		writeAuthFailure(w, manager.ErrAuthenticationFail)
		return
	}

	cluster, found := manager.Default.LookupCluster(hisAppid)
	if !found {
		log.Error("create job %s(%s) {appid:%s topic:%s ver:%s} invalid appid",
			r.RemoteAddr, realIp, hisAppid, topic, ver)

		writeBadRequest(w, "invalid appid")
		return
	}

	log.Info("create job[%s] %s(%s) {appid:%s topic:%s ver:%s}",
		appid, r.RemoteAddr, realIp, hisAppid, topic, ver)

	rawTopic := manager.Default.KafkaTopic(hisAppid, topic, ver)
	if err := job.Default.CreateJobQueue(Options.AssignJobShardId, hisAppid, rawTopic); err != nil {
		log.Error("create job[%s] %s(%s) {shard:%d appid:%s topic:%s ver:%s} %v",
			appid, r.RemoteAddr, realIp, Options.AssignJobShardId, hisAppid, topic, ver, err)

		writeServerError(w, err.Error())
		return
	}

	if err := this.gw.zkzone.CreateJobQueue(rawTopic, cluster); err != nil {
		log.Error("app[%s] %s(%s) create job: {shard:%d appid:%s topic:%s ver:%s} %v",
			appid, r.RemoteAddr, realIp, Options.AssignJobShardId, hisAppid, topic, ver, err)

		writeServerError(w, err.Error())
		return
	}

	w.WriteHeader(http.StatusCreated)
	w.Write(ResponseOk)
}
开发者ID:funkygao，项目名称:gafka，代码行数:61，代码来源:handler_man.go


示例4: schemaHandler

//go:generate goannotation $GOFILE
// @rest GET /v1/schema/:appid/:topic/:ver
func (this *manServer) schemaHandler(w http.ResponseWriter, r *http.Request, params httprouter.Params) {
	hisAppid := params.ByName(UrlParamAppid)
	myAppid := r.Header.Get(HttpHeaderAppid)
	topic := params.ByName(UrlParamTopic)
	ver := params.ByName(UrlParamVersion)
	realIp := getHttpRemoteIp(r)

	log.Info("schema[%s] %s(%s) {app:%s topic:%s ver:%s UA:%s}",
		myAppid, r.RemoteAddr, realIp, hisAppid, topic, ver, r.Header.Get("User-Agent"))

	// TODO authorization

	_, found := manager.Default.LookupCluster(hisAppid)
	if !found {
		writeBadRequest(w, "invalid appid")
		return
	}

	// TODO lookup from manager and send reponse
	schema, err := manager.Default.TopicSchema(hisAppid, topic, ver)
	if err != nil {
		writeBadRequest(w, err.Error())
		return
	}

	w.Write([]byte(strings.TrimSpace(schema)))
}
开发者ID:funkygao，项目名称:gafka，代码行数:29，代码来源:handler_man.go


示例5: partitionsHandler

// @rest GET /v1/partitions/:appid/:topic/:ver
func (this *manServer) partitionsHandler(w http.ResponseWriter, r *http.Request, params httprouter.Params) {
	topic := params.ByName(UrlParamTopic)
	hisAppid := params.ByName(UrlParamAppid)
	appid := r.Header.Get(HttpHeaderAppid)
	pubkey := r.Header.Get(HttpHeaderPubkey)
	ver := params.ByName(UrlParamVersion)
	realIp := getHttpRemoteIp(r)

	cluster, found := manager.Default.LookupCluster(hisAppid)
	if !found {
		log.Error("partitions[%s] %s(%s) {app:%s topic:%s ver:%s} invalid appid",
			appid, r.RemoteAddr, realIp, hisAppid, topic, ver)

		writeBadRequest(w, "invalid appid")
		return
	}

	if !manager.Default.AuthAdmin(appid, pubkey) {
		log.Warn("suspicous partitions call from %s(%s) {cluster:%s app:%s key:%s topic:%s ver:%s}",
			r.RemoteAddr, realIp, cluster, appid, pubkey, topic, ver)

		writeAuthFailure(w, manager.ErrAuthenticationFail)
		return
	}

	log.Info("partitions[%s] %s(%s) {cluster:%s app:%s topic:%s ver:%s}",
		appid, r.RemoteAddr, realIp, cluster, hisAppid, topic, ver)

	zkcluster := meta.Default.ZkCluster(cluster)
	if zkcluster == nil {
		log.Error("suspicous partitions call from %s(%s) {cluster:%s app:%s key:%s topic:%s ver:%s} undefined cluster",
			r.RemoteAddr, realIp, cluster, appid, pubkey, topic, ver)

		writeBadRequest(w, "undefined cluster")
		return
	}

	kfk, err := sarama.NewClient(zkcluster.BrokerList(), sarama.NewConfig())
	if err != nil {
		log.Error("cluster[%s] %v", zkcluster.Name(), err)

		writeServerError(w, err.Error())
		return
	}
	defer kfk.Close()

	partitions, err := kfk.Partitions(manager.Default.KafkaTopic(hisAppid, topic, ver))
	if err != nil {
		log.Error("cluster[%s] from %s(%s) {app:%s topic:%s ver:%s} %v",
			zkcluster.Name(), r.RemoteAddr, realIp, hisAppid, topic, ver, err)

		writeServerError(w, err.Error())
		return
	}

	w.Write([]byte(fmt.Sprintf(`{"num": %d}`, len(partitions))))
}
开发者ID:funkygao，项目名称:gafka，代码行数:58，代码来源:handler_man.go


示例6: deleteWebhookHandler

// @rest DELETE /v1/jobs/:appid/:topic/:ver?group=xx
func (this *manServer) deleteWebhookHandler(w http.ResponseWriter, r *http.Request, params httprouter.Params) {
	topic := params.ByName(UrlParamTopic)
	if !manager.Default.ValidateTopicName(topic) {
		log.Warn("illegal topic: %s", topic)

		writeBadRequest(w, "illegal topic")
		return
	}

	query := r.URL.Query()
	group := query.Get("group")
	realIp := getHttpRemoteIp(r)
	hisAppid := params.ByName(UrlParamAppid)
	myAppid := r.Header.Get(HttpHeaderAppid)
	ver := params.ByName(UrlParamVersion)

	if err := manager.Default.AuthSub(myAppid, r.Header.Get(HttpHeaderSubkey),
		hisAppid, topic, group); err != nil {
		log.Error("+webhook[%s/%s] -(%s): {%s.%s.%s UA:%s} %v",
			myAppid, group, realIp, hisAppid, topic, ver, r.Header.Get("User-Agent"), err)

		writeAuthFailure(w, err)
		return
	}

	/*
		cluster, found := manager.Default.LookupCluster(hisAppid)
		if !found {
			log.Error("+webhook[%s/%s] -(%s): {%s.%s.%s UA:%s} undefined cluster",
				myAppid, group, realIp, hisAppid, topic, ver, r.Header.Get("User-Agent"))

			writeBadRequest(w, "invalid appid")
			return
		}

		log.Info("+webhook[%s/%s] %s(%s): {%s.%s.%s UA:%s}",
			myAppid, group, r.RemoteAddr, realIp, hisAppid, topic, ver, r.Header.Get("User-Agent"))

		rawTopic := manager.Default.KafkaTopic(hisAppid, topic, ver)
	*/

}
开发者ID:funkygao，项目名称:gafka，代码行数:43，代码来源:handler_man.go


示例7: setOptionHandler

// @rest PUT /v1/options/:option/:value
func (this *manServer) setOptionHandler(w http.ResponseWriter, r *http.Request, params httprouter.Params) {
	option := params.ByName("option")
	value := params.ByName("value")
	boolVal := value == "true"

	// TODO auth

	switch option {
	case "debug":
		Options.Debug = boolVal

	case "nometrics":
		Options.DisableMetrics = boolVal

	case "gzip":
		Options.EnableGzip = boolVal

	case "badgroup_rater":
		Options.BadGroupRateLimit = boolVal

	case "badpub_rater":
		Options.BadPubAppRateLimit = boolVal

	case "refreshdb":
		manager.Default.ForceRefresh()

	case "ratelimit":
		Options.Ratelimit = boolVal

	case "resethh":
		hh.Default.ResetCounters()

	case "hh":
		Options.EnableHintedHandoff = boolVal
		if !boolVal {
			hh.Default.Stop()
			w.Write([]byte(fmt.Sprintf("id:%s hh[%s] stopped", Options.Id, hh.Default.Name())))
			return
		} else {
			if err := hh.Default.Start(); err != nil {
				writeServerError(w, err.Error())
				return
			} else {
				w.Write([]byte(fmt.Sprintf("id:%s hh[%s] started", Options.Id, hh.Default.Name())))
				return
			}
		}

	case "hhflush":
		if boolVal {
			if hh.Default == nil {
				writeServerError(w, "no underlying hinted handoff")
			} else if Options.EnableHintedHandoff {
				writeBadRequest(w, "turn off hinted handoff first")
			} else {
				hh.Default.FlushInflights()
				w.Write([]byte(fmt.Sprintf("id:%s hh[%s] inflights flushed", Options.Id, hh.Default.Name())))
			}
			return
		}

	case "jobshardid":
		shardId, err := strconv.Atoi(value)
		if err != nil {
			writeBadRequest(w, "invalid job shard id")
			return
		} else {
			Options.AssignJobShardId = shardId
		}

	case "punish":
		d, err := time.ParseDuration(value)
		if err != nil {
			writeBadRequest(w, err.Error())
			return
		} else {
			Options.BadClientPunishDuration = d
		}

	case "500backoff":
		d, err := time.ParseDuration(value)
		if err != nil {
			writeBadRequest(w, err.Error())
			return
		} else {
			Options.InternalServerErrorBackoff = d
		}

	case "auditpub":
		Options.AuditPub = boolVal

	case "auditsub":
		Options.AuditSub = boolVal

	case "allhh":
		Options.AllwaysHintedHandoff = boolVal

	case "standbysub":
		Options.PermitStandbySub = boolVal
//.........这里部分代码省略.........
开发者ID:funkygao，项目名称:gafka，代码行数:101，代码来源:handler_man.go


示例8: alterTopicHandler

// @rest PUT /v1/topics/:appid/:topic/:ver?partitions=1&retention.hours=72&retention.bytes=-1
func (this *manServer) alterTopicHandler(w http.ResponseWriter, r *http.Request, params httprouter.Params) {
	topic := params.ByName(UrlParamTopic)
	if !manager.Default.ValidateTopicName(topic) {
		log.Warn("illegal topic: %s", topic)

		writeBadRequest(w, "illegal topic")
		return
	}

	realIp := getHttpRemoteIp(r)

	if !this.throttleAddTopic.Pour(realIp, 1) {
		writeQuotaExceeded(w)
		return
	}

	hisAppid := params.ByName(UrlParamAppid)
	appid := r.Header.Get(HttpHeaderAppid)
	pubkey := r.Header.Get(HttpHeaderPubkey)
	ver := params.ByName(UrlParamVersion)
	if !manager.Default.AuthAdmin(appid, pubkey) {
		log.Warn("suspicous alter topic from %s(%s) {appid:%s pubkey:%s topic:%s ver:%s}",
			r.RemoteAddr, realIp, appid, pubkey, topic, ver)

		writeAuthFailure(w, manager.ErrAuthenticationFail)
		return
	}

	cluster, found := manager.Default.LookupCluster(hisAppid)
	if !found {
		log.Error("alter topic[%s] %s(%s) {app:%s topic:%s ver:%s} invalid appid",
			appid, r.RemoteAddr, realIp, hisAppid, topic, ver)

		writeBadRequest(w, "invalid appid")
		return
	}

	zkcluster := meta.Default.ZkCluster(cluster)
	if zkcluster == nil {
		log.Error("alter topic from %s(%s) {appid:%s pubkey:%s cluster:%s topic:%s ver:%s} undefined cluster",
			r.RemoteAddr, realIp, appid, pubkey, cluster, topic, ver)

		writeBadRequest(w, "undefined cluster")
		return
	}

	info := zkcluster.RegisteredInfo()
	if !info.Public {
		log.Warn("app[%s] alter topic:%s in non-public cluster: %+v", hisAppid, topic, params)

		writeBadRequest(w, "invalid cluster")
		return
	}

	ts := sla.DefaultSla()
	query := r.URL.Query()
	if partitionsArg := query.Get(sla.SlaKeyPartitions); partitionsArg != "" {
		ts.Partitions, _ = strconv.Atoi(partitionsArg)
	}
	if retentionBytes := query.Get(sla.SlaKeyRetentionBytes); retentionBytes != "" {
		ts.RetentionBytes, _ = strconv.Atoi(retentionBytes)
	}
	ts.ParseRetentionHours(query.Get(sla.SlaKeyRetentionHours))

	// validate the sla
	if err := ts.Validate(); err != nil {
		log.Error("app[%s] alter topic:%s %s: %+v", hisAppid, topic, query.Encode(), err)

		writeBadRequest(w, err.Error())
		return
	}

	log.Info("app[%s] from %s(%s) alter topic: {appid:%s cluster:%s topic:%s ver:%s query:%s}",
		appid, r.RemoteAddr, realIp, hisAppid, cluster, topic, ver, query.Encode())

	rawTopic := manager.Default.KafkaTopic(hisAppid, topic, ver)
	alterConfig := ts.DumpForAlterTopic()
	if len(alterConfig) == 0 {
		log.Warn("app[%s] from %s(%s) alter topic: {appid:%s cluster:%s topic:%s ver:%s query:%s} nothing updated",
			appid, r.RemoteAddr, realIp, hisAppid, cluster, topic, ver, query.Encode())

		writeBadRequest(w, "nothing updated")
		return
	}

	lines, err := zkcluster.AlterTopic(rawTopic, ts)
	if err != nil {
		log.Error("app[%s] from %s(%s) alter topic: {appid:%s cluster:%s topic:%s ver:%s query:%s} %v",
			appid, r.RemoteAddr, realIp, hisAppid, cluster, topic, ver, query.Encode(), err)

		writeServerError(w, err.Error())
		return
	}

	for _, l := range lines {
		log.Trace("app[%s] alter topic[%s] in cluster %s: %s", appid, rawTopic, cluster, l)
	}

	w.Write(ResponseOk)
//.........这里部分代码省略.........
开发者ID:funkygao，项目名称:gafka，代码行数:101，代码来源:handler_man.go


示例9: createWebhookHandler

// @rest PUT /v1/webhook/:appid/:topic/:ver?group=xx
func (this *manServer) createWebhookHandler(w http.ResponseWriter, r *http.Request, params httprouter.Params) {
	topic := params.ByName(UrlParamTopic)
	if !manager.Default.ValidateTopicName(topic) {
		log.Warn("illegal topic: %s", topic)

		writeBadRequest(w, "illegal topic")
		return
	}

	query := r.URL.Query()
	group := query.Get("group")
	realIp := getHttpRemoteIp(r)
	hisAppid := params.ByName(UrlParamAppid)
	myAppid := r.Header.Get(HttpHeaderAppid)
	ver := params.ByName(UrlParamVersion)

	if err := manager.Default.AuthSub(myAppid, r.Header.Get(HttpHeaderSubkey),
		hisAppid, topic, group); err != nil {
		log.Error("+webhook[%s/%s] -(%s): {%s.%s.%s UA:%s} %v",
			myAppid, group, realIp, hisAppid, topic, ver, r.Header.Get("User-Agent"), err)

		writeAuthFailure(w, err)
		return
	}

	cluster, found := manager.Default.LookupCluster(hisAppid)
	if !found {
		log.Error("+webhook[%s/%s] -(%s): {%s.%s.%s UA:%s} undefined cluster",
			myAppid, group, realIp, hisAppid, topic, ver, r.Header.Get("User-Agent"))

		writeBadRequest(w, "invalid appid")
		return
	}

	log.Info("+webhook[%s/%s] %s(%s): {%s.%s.%s UA:%s}",
		myAppid, group, r.RemoteAddr, realIp, hisAppid, topic, ver, r.Header.Get("User-Agent"))

	rawTopic := manager.Default.KafkaTopic(hisAppid, topic, ver)
	var hook zk.WebhookMeta
	decoder := json.NewDecoder(r.Body)
	if err := decoder.Decode(&hook); err != nil {
		log.Error("+webhook[%s/%s] %s(%s): {%s.%s.%s UA:%s} %v",
			myAppid, group, r.RemoteAddr, realIp, hisAppid, topic, ver, r.Header.Get("User-Agent"), err)

		writeBadRequest(w, err.Error())
		return
	}
	r.Body.Close()

	// validate the url
	for _, ep := range hook.Endpoints {
		_, err := url.ParseRequestURI(ep)
		if err != nil {
			log.Error("+webhook[%s/%s] %s(%s): {%s.%s.%s UA:%s} %+v %v",
				myAppid, group, r.RemoteAddr, realIp, hisAppid, topic, ver, r.Header.Get("User-Agent"), hook.Endpoints, err)

			writeBadRequest(w, err.Error())
			return
		}
	}

	hook.Cluster = cluster // cluster is decided by server
	if err := this.gw.zkzone.CreateOrUpdateWebhook(rawTopic, hook); err != nil {
		log.Error("+webhook[%s/%s] %s(%s): {%s.%s.%s UA:%s} %v",
			myAppid, group, r.RemoteAddr, realIp, hisAppid, topic, ver, r.Header.Get("User-Agent"), err)

		writeServerError(w, err.Error())
		return
	}

	w.Write(ResponseOk)
}
开发者ID:funkygao，项目名称:gafka，代码行数:73，代码来源:handler_man.go


示例10: ackHandler

//go:generate goannotation $GOFILE
// @rest PUT /v1/offsets/:appid/:topic/:ver/:group with json body
func (this *subServer) ackHandler(w http.ResponseWriter, r *http.Request, params httprouter.Params) {
	var (
		topic    string
		ver      string
		myAppid  string
		hisAppid string
		group    string
		err      error
	)

	group = params.ByName(UrlParamGroup)
	ver = params.ByName(UrlParamVersion)
	topic = params.ByName(UrlParamTopic)
	hisAppid = params.ByName(UrlParamAppid)
	myAppid = r.Header.Get(HttpHeaderAppid)

	if err = manager.Default.AuthSub(myAppid, r.Header.Get(HttpHeaderSubkey),
		hisAppid, topic, group); err != nil {
		writeAuthFailure(w, err)
		return
	}

	cluster, found := manager.Default.LookupCluster(hisAppid)
	if !found {
		writeBadRequest(w, "invalid appid")
		return
	}

	msgLen := int(r.ContentLength)
	switch {
	case int64(msgLen) > Options.MaxPubSize:
		writeBadRequest(w, ErrTooBigMessage.Error())
		return

	case msgLen < Options.MinPubSize:
		writeBadRequest(w, ErrTooSmallMessage.Error())
		return
	}

	var msg *mpool.Message
	msg = mpool.NewMessage(msgLen)
	msg.Body = msg.Body[0:msgLen]
	lbr := io.LimitReader(r.Body, Options.MaxPubSize+1)
	if _, err = io.ReadAtLeast(lbr, msg.Body, msgLen); err != nil {
		msg.Free()

		writeBadRequest(w, ErrTooBigMessage.Error())
		return
	}

	var acks ackOffsets
	if err = json.Unmarshal(msg.Body, &acks); err != nil {
		msg.Free()

		writeBadRequest(w, "invalid ack json body")
		return
	}

	msg.Free()

	realIp := getHttpRemoteIp(r)
	realGroup := myAppid + "." + group
	rawTopic := manager.Default.KafkaTopic(hisAppid, topic, ver)
	for i := 0; i < len(acks); i++ {
		acks[i].cluster = cluster
		acks[i].topic = rawTopic
		acks[i].group = realGroup
	}

	log.Debug("ack[%s/%s] %s(%s) {%s.%s.%s UA:%s} %+v",
		myAppid, group, r.RemoteAddr, realIp, hisAppid, topic, ver, r.Header.Get("User-Agent"), acks)

	if atomic.AddInt32(&this.ackShutdown, 1) == 0 {
		// kateway is shutting down, ackCh is already closed
		log.Warn("ack[%s/%s] %s(%s) {%s.%s.%s UA:%s} server is shutting down %+v ",
			myAppid, group, r.RemoteAddr, realIp, hisAppid, topic, ver, r.Header.Get("User-Agent"), acks)

		writeServerError(w, "server is shutting down")
		return
	}

	this.ackCh <- acks
	atomic.AddInt32(&this.ackShutdown, -1)

	w.Write(ResponseOk)
}
开发者ID:funkygao，项目名称:gafka，代码行数:88，代码来源:handler_sub_ack.go


示例11: subRawHandler

//go:generate goannotation $GOFILE
// @rest GET /v1/raw/msgs/:cluster/:topic?group=xx&batch=10&reset=<newest|oldest>
func (this *subServer) subRawHandler(w http.ResponseWriter, r *http.Request, params httprouter.Params) {
	var (
		cluster string
		topic   string
		myAppid string
		reset   string
		group   string
		limit   int // max messages to include in the message set
		err     error
	)

	if !Options.DisableMetrics {
		this.subMetrics.SubTryQps.Mark(1)
	}

	query := r.URL.Query()
	group = query.Get("group")
	reset = query.Get("reset")
	realIp := getHttpRemoteIp(r)

	if !manager.Default.ValidateGroupName(r.Header, group) {
		log.Error("sub raw -(%s): illegal group: %s", realIp, group)
		this.subMetrics.ClientError.Mark(1)
		writeBadRequest(w, "illegal group")
		return
	}

	limit, err = getHttpQueryInt(&query, "batch", 1)
	if err != nil {
		log.Error("sub raw -(%s): illegal batch: %v", realIp, err)
		this.subMetrics.ClientError.Mark(1)
		writeBadRequest(w, "illegal batch")
		return
	}
	if limit > Options.MaxSubBatchSize && Options.MaxSubBatchSize > 0 {
		limit = Options.MaxSubBatchSize
	}

	topic = params.ByName(UrlParamTopic)
	cluster = params.ByName("cluster")
	myAppid = r.Header.Get(HttpHeaderAppid)

	log.Debug("sub raw[%s/%s] %s(%s) {%s/%s batch:%d UA:%s}",
		myAppid, group, r.RemoteAddr, realIp, cluster, topic, limit, r.Header.Get("User-Agent"))

	if !Options.DisableMetrics {
		this.subMetrics.SubQps.Mark(1)
	}

	fetcher, err := store.DefaultSubStore.Fetch(cluster, topic,
		myAppid+"."+group, r.RemoteAddr, realIp, reset, Options.PermitStandbySub)
	if err != nil {
		// e,g. kafka was totally shutdown
		// e,g. too many consumers for the same group
		log.Error("sub raw[%s/%s] %s(%s) {%s/%s batch:%d UA:%s} %v",
			myAppid, group, r.RemoteAddr, realIp, cluster, topic, limit, r.Header.Get("User-Agent"), err)

		if store.DefaultSubStore.IsSystemError(err) {
			writeServerError(w, err.Error())
		} else {
			writeBadRequest(w, err.Error())
		}

		return
	}

	var gz *gzip.Writer
	w, gz = gzipWriter(w, r)
	err = this.pumpRawMessages(w, r, realIp, fetcher, limit, myAppid, topic, group)
	if err != nil {
		// e,g. broken pipe, io timeout, client gone
		// e,g. kafka: error while consuming app1.foobar.v1/0: EOF (kafka was shutdown)
		log.Error("sub raw[%s/%s] %s(%s) {%s/%s batch:%d UA:%s} %v",
			myAppid, group, r.RemoteAddr, realIp, cluster, topic, limit, r.Header.Get("User-Agent"), err)

		if err != ErrClientGone {
			if store.DefaultSubStore.IsSystemError(err) {
				writeServerError(w, err.Error())
			} else {
				writeBadRequest(w, err.Error())
			}
		}

		// fetch.Close might be called by subServer.closedConnCh
		if err = fetcher.Close(); err != nil {
			log.Error("sub raw[%s/%s] %s(%s) {%s/%s batch:%d UA:%s} %v",
				myAppid, group, r.RemoteAddr, realIp, cluster, topic, limit, r.Header.Get("User-Agent"), err)
		}
	}

	if gz != nil {
		gz.Close()
	}
}
开发者ID:funkygao，项目名称:gafka，代码行数:96，代码来源:handler_sub_raw.go


示例12: subHandler

//go:generate goannotation $GOFILE
// @rest GET /v1/msgs/:appid/:topic/:ver?group=xx&batch=10&reset=<newest|oldest>&ack=1&q=<dead|retry>
func (this *subServer) subHandler(w http.ResponseWriter, r *http.Request, params httprouter.Params) {
	var (
		topic      string
		ver        string
		myAppid    string
		hisAppid   string
		reset      string
		group      string
		realGroup  string
		shadow     string
		rawTopic   string
		partition  string
		partitionN int = -1
		offset     string
		offsetN    int64 = -1
		limit      int   // max messages to include in the message set
		delayedAck bool  // last acked partition/offset piggybacked on this request
		err        error
	)

	if !Options.DisableMetrics {
		this.subMetrics.SubTryQps.Mark(1)
	}

	query := r.URL.Query()
	group = query.Get("group")
	myAppid = r.Header.Get(HttpHeaderAppid)
	realGroup = myAppid + "." + group
	reset = query.Get("reset")
	realIp := getHttpRemoteIp(r)

	if !manager.Default.ValidateGroupName(r.Header, group) {
		log.Error("sub -(%s): illegal group: %s", realIp, group)
		this.subMetrics.ClientError.Mark(1)
		writeBadRequest(w, "illegal group")
		return
	}

	if Options.BadGroupRateLimit && !this.throttleBadGroup.Pour(realGroup, 0) {
		this.goodGroupLock.RLock()
		_, good := this.goodGroupClients[r.RemoteAddr]
		this.goodGroupLock.RUnlock()

		if !good {
			// this bad group client is in confinement period
			log.Error("sub -(%s): group[%s] failure quota exceeded %s", realIp, realGroup, r.Header.Get("User-Agent"))
			writeQuotaExceeded(w)
			return
		}
	}

	limit, err = getHttpQueryInt(&query, "batch", 1)
	if err != nil {
		log.Error("sub -(%s): illegal batch: %v", realIp, err)
		this.subMetrics.ClientError.Mark(1)
		writeBadRequest(w, "illegal batch")
		return
	}
	if limit > Options.MaxSubBatchSize && Options.MaxSubBatchSize > 0 {
		limit = Options.MaxSubBatchSize
	}

	ver = params.ByName(UrlParamVersion)
	topic = params.ByName(UrlParamTopic)
	hisAppid = params.ByName(UrlParamAppid)

	// auth
	if err = manager.Default.AuthSub(myAppid, r.Header.Get(HttpHeaderSubkey),
		hisAppid, topic, group); err != nil {
		log.Error("sub[%s/%s] -(%s): {%s.%s.%s UA:%s} %v",
			myAppid, group, realIp, hisAppid, topic, ver, r.Header.Get("User-Agent"), err)

		this.subMetrics.ClientError.Mark(1)
		writeAuthFailure(w, err)
		return
	}

	// fetch the client ack partition and offset
	delayedAck = query.Get("ack") == "1"
	if delayedAck {
		// consumers use explicit acknowledges in order to signal a message as processed successfully
		// if consumers fail to ACK, the message hangs and server will refuse to move ahead

		// get the partitionN and offsetN from client header
		// client will ack with partition=-1, offset=-1:
		// 1. handshake phase
		// 2. when 204 No Content
		partition = r.Header.Get(HttpHeaderPartition)
		offset = r.Header.Get(HttpHeaderOffset)
		if partition != "" && offset != "" {
			// convert partition and offset to int
			offsetN, err = strconv.ParseInt(offset, 10, 64)
			if err != nil {
				log.Error("sub[%s/%s] %s(%s) {%s.%s.%s UA:%s} offset:%s",
					myAppid, group, r.RemoteAddr, realIp, hisAppid, topic, ver, r.Header.Get("User-Agent"), offset)

				this.subMetrics.ClientError.Mark(1)
				writeBadRequest(w, "ack with bad offset")
//.........这里部分代码省略.........
开发者ID:funkygao，项目名称:gafka，代码行数:101，代码来源:handler_sub.go


示例13: pubRawHandler

//go:generate goannotation $GOFILE
// @rest POST /v1/raw/msgs/:cluster/:topic?key=mykey&async=1&ack=all
func (this *pubServer) pubRawHandler(w http.ResponseWriter, r *http.Request, params httprouter.Params) {
	var (
		cluster      string
		topic        string
		partitionKey string
		t1           = time.Now()
	)

	if !Options.DisableMetrics {
		this.pubMetrics.PubTryQps.Mark(1)
	}

	realIp := getHttpRemoteIp(r)
	topic = params.ByName(UrlParamTopic)
	cluster = params.ByName("cluster")

	buf := bytes.NewBuffer(make([]byte, 0, 1<<10))
	_, err := buf.ReadFrom(r.Body)
	if err != nil {
		log.Error("pub raw %s(%s) {C:%s T:%s UA:%s} %s",
			r.RemoteAddr, realIp, cluster, topic, r.Header.Get("User-Agent"), err)

		this.pubMetrics.ClientError.Inc(1)
		writeBadRequest(w, err.Error())
		return
	}

	body := buf.Bytes()

	if !Options.DisableMetrics {
		this.pubMetrics.PubQps.Mark(1)
		this.pubMetrics.PubMsgSize.Update(int64(len(body)))
	}

	query := r.URL.Query() // reuse the query will save 100ns
	partitionKey = query.Get("key")

	pubMethod := store.DefaultPubStore.SyncPub
	if query.Get("async") == "1" {
		pubMethod = store.DefaultPubStore.AsyncPub
	}
	if query.Get("ack") == "all" {
		pubMethod = store.DefaultPubStore.SyncAllPub
	}

	_, _, err = pubMethod(cluster, topic, []byte(partitionKey), body)
	if err != nil {
		log.Error("pub raw %s(%s) {C:%s T:%s UA:%s} %s",
			r.RemoteAddr, realIp, cluster, topic, r.Header.Get("User-Agent"), err)

		writeServerError(w, err.Error())
		return
	}

	w.WriteHeader(http.StatusCreated)

	if _, err = w.Write(ResponseOk); err != nil {
		log.Error("pub raw %s(%s) {C:%s T:%s UA:%s} %s",
			r.RemoteAddr, realIp, cluster, topic, r.Header.Get("User-Agent"), err)

		this.pubMetrics.ClientError.Inc(1)
	}

	if !Options.DisableMetrics {
		this.pubMetrics.PubLatency.Update(time.Since(t1).Nanoseconds() / 1e6) // in ms
	}

}
开发者ID:funkygao，项目名称:gafka，代码行数:70，代码来源:handler_pub_raw.go


示例14: buryHandler

//go:generate goannotation $GOFILE
// @rest PUT /v1/msgs/:appid/:topic/:ver?group=xx&q=<dead|retry>
// q=retry&X-Bury=dead means bury from retry queue to dead queue
func (this *subServer) buryHandler(w http.ResponseWriter, r *http.Request, params httprouter.Params) {
	var (
		topic      string
		ver        string
		myAppid    string
		hisAppid   string
		group      string
		rawTopic   string
		shadow     string
		bury       string
		partition  string
		partitionN int = -1
		offset     string
		offsetN    int64 = -1
		err        error
	)

	query := r.URL.Query()
	group = query.Get("group")
	if !manager.Default.ValidateGroupName(r.Header, group) {
		writeBadRequest(w, "illegal group")
		return
	}

	ver = params.ByName(UrlParamVersion)
	topic = params.ByName(UrlParamTopic)
	hisAppid = params.ByName(UrlParamAppid)
	myAppid = r.Header.Get(HttpHeaderAppid)
	realIp := getHttpRemoteIp(r)

	bury = r.Header.Get(HttpHeaderMsgBury)
	if !sla.ValidateShadowName(bury) {
		log.Error("bury[%s/%s] %s(%s) {%s.%s.%s UA:%s} illegal bury: %s",
			myAppid, group, r.RemoteAddr, realIp, hisAppid, topic, ver, r.Header.Get("User-Agent"), bury)

		writeBadRequest(w, "illegal bury")
		return
	}

	// auth
	if err = manager.Default.AuthSub(myAppid, r.Header.Get(HttpHeaderSubkey),
		hisAppid, topic, group); err != nil {
		log.Error("bury[%s/%s] %s(%s) {%s.%s.%s UA:%s} %v",
			myAppid, group, r.RemoteAddr, realIp, hisAppid, topic, ver, r.Header.Get("User-Agent"), err)

		writeAuthFailure(w, err)
		return
	}

	partition = r.Header.Get(HttpHeaderPartition)
	offset = r.Header.Get(HttpHeaderOffset)
	if partition == "" || offset == "" {
		log.Error("bury[%s/%s] %s(%s) {%s.%s.%s UA:%s} empty offset or partition",
			myAppid, group, r.RemoteAddr, realIp, hisAppid, topic, ver, r.Header.Get("User-Agent"))

		writeBadRequest(w, "empty offset or partition")
		return
	}

	offsetN, err = strconv.ParseInt(offset, 10, 64)
	if err != nil || offsetN < 0 {
		log.Error("bury[%s/%s] %s(%s) {%s.%s.%s UA:%s} illegal offset:%s",
			myAppid, group, r.RemoteAddr, realIp, hisAppid, topic, ver, r.Header.Get("User-Agent"), offset)

		writeBadRequest(w, "bad offset")
		return
	}
	partitionN, err = strconv.Atoi(partition)
	if err != nil || partitionN < 0 {
		log.Error("bury[%s/%s] %s(%s) {%s.%s.%s UA:%s} illegal partition:%s",
			myAppid, group, r.RemoteAddr, realIp, hisAppid, topic, ver, r.Header.Get("User-Agent"), partition)

		writeBadRequest(w, "bad partition")
		return
	}

	shadow = query.Get("q")

	log.Debug("bury[%s/%s] %s(%s) {%s.%s.%s bury:%s shadow=%s partition:%s offset:%s UA:%s}",
		myAppid, group, r.RemoteAddr, realIp, hisAppid, topic, ver, bury, shadow, partition, offset, r.Header.Get("User-Agent"))

	msgLen := int(r.ContentLength)
	msg := make([]byte, msgLen)
	if _, err = io.ReadAtLeast(r.Body, msg, msgLen); err != nil {
		log.Error("bury[%s/%s] %s(%s) {%s.%s.%s UA:%s} %v",
			myAppid, group, r.RemoteAddr, realIp, hisAppid, topic, ver, r.Header.Get("User-Agent"), err)

		writeBadRequest(w, err.Error())
		return
	}

	cluster, found := manager.Default.LookupCluster(hisAppid)
	if !found {
		log.Error("bury[%s/%s] %s(%s) {%s.%s.%s UA:%s} invalid appid:%s",
			myAppid, group, r.RemoteAddr, realIp, hisAppid, topic, ver, r.Header.Get("User-Agent"), hisAppid)

		writeBadRequest(w, "invalid appid")
//.........这里部分代码省略.........
开发者ID:funkygao，项目名称:gafka，代码行数:101，代码来源:handler_sub_bury.go


示例15: pubHandler

//go:generate goannotation $GOFILE
// @rest POST /v1/msgs/:topic/:ver?key=mykey&async=1&ack=all&hh=n
func (this *pubServer) pubHandler(w http.ResponseWriter, r *http.Request, params httprouter.Params) {
	var (
		appid        string
		topic        string
		ver          string
		tag          string
		partitionKey string
		async        bool
		hhDisabled   bool // hh enabled by default
		t1           = time.Now()
	)

	if !Options.DisableMetrics {
		this.pubMetrics.PubTryQps.Mark(1)
	}

	realIp := getHttpRemoteIp(r)
	if Options.Ratelimit && !this.throttlePub.Pour(realIp, 1) {
		log.Warn("pub[%s] %s(%s) rate limit reached: %d/s", appid, r.RemoteAddr, realIp, Options.PubQpsLimit)

		this.pubMetrics.ClientError.Inc(1)
		writeQuotaExceeded(w)
		return
	}

	appid = r.Header.Get(HttpHeaderAppid)
	topic = params.ByName(UrlParamTopic)
	ver = params.ByName(UrlParamVersion)

	if err := manager.Default.OwnTopic(appid, r.Header.Get(HttpHeaderPubkey), topic); err != nil {
		log.Warn("pub[%s] %s(%s) {topic:%s ver:%s UA:%s} %s",
			appid, r.RemoteAddr, realIp, topic, ver, r.Header.Get("User-Agent"), err)

		this.pubMetrics.ClientError.Inc(1)
		this.respond4XX(appid, w, err.Error(), http.StatusUnauthorized)
		return
	}

	msgLen := int(r.ContentLength)
	switch {
	case int64(msgLen) > Options.MaxPubSize:
		log.Warn("pub[%s] %s(%s) {topic:%s ver:%s UA:%s} too big content length: %d",
			appid, r.RemoteAddr, realIp, topic, ver, r.Header.Get("User-Agent"), msgLen)

		this.pubMetrics.ClientError.Inc(1)
		this.respond4XX(appid, w, ErrTooBigMessage.Error(), http.StatusBadRequest)
		return

	case msgLen < Options.MinPubSize:
		log.Warn("pub[%s] %s(%s) {topic:%s ver:%s UA:%s} too small content length: %d",
			appid, r.RemoteAddr, realIp, topic, ver, r.Header.Get("User-Agent"), msgLen)

		this.pubMetrics.ClientError.Inc(1)
		this.respond4XX(appid, w, ErrTooSmallMessage.Error(), http.StatusBadRequest)
		return
	}

	query := r.URL.Query() // reuse the query will save 100ns

	partitionKey = query.Get("key")
	if len(partitionKey) > MaxPartitionKeyLen {
		log.Warn("pub[%s] %s(%s) {topic:%s ver:%s UA:%s} too big key: %s",
			appid, r.RemoteAddr, realIp, topic, ver,
			r.Header.Get("User-Agent"), partitionKey)

		this.pubMetrics.ClientError.Inc(1)
		this.respond4XX(appid, w, "too big key", http.StatusBadRequest)
		return
	}

	var msg *mpool.Message
	tag = r.Header.Get(HttpHeaderMsgTag)
	if tag != "" {
		if len(tag) > Options.MaxMsgTagLen {
			this.respond4XX(appid, w, "too big tag", http.StatusBadRequest)
			return
		}

		msgSz := tagLen(tag) + msgLen
		msg = mpool.NewMessage(msgSz)
		msg.Body = msg.Body[0:msgSz]
	} else {
		msg = mpool.NewMessage(msgLen)
		msg.Body = msg.Body[0:msgLen]
	}

	// get the raw POST message, if body more than content-length ignore the extra payload
	lbr := io.LimitReader(r.Body, Options.MaxPubSize+1)
	if _, err := io.ReadAtLeast(lbr, msg.Body, msgLen); err != nil {
		msg.Free()

		log.Error("pub[%s] %s(%s) {topic:%s ver:%s UA:%s} %s",
			appid, r.RemoteAddr, realIp, topic, ver, r.Header.Get("User-Agent"), err)

		this.pubMetrics.ClientError.Inc(1)
		this.respond4XX(appid, w, err.Error(), http.StatusBadRequest) // TODO http.StatusRequestEntityTooLarge
		return
	}
//.........这里部分代码省略.........
开发者ID:funkygao，项目名称:gafka，代码行数:101，代码来源:handler_pub.go


示例16: addJobHandler

//go:generate goannotation $GOFILE
// @rest POST /v1/jobs/:topic/:ver?delay=100|due=1471565204
// TODO tag, partitionKey
// TODO use dedicated metrics
func (this *pubServer) addJobHandler(w http.ResponseWriter, r *http.Request, params httprouter.Params) {
	if !Options.DisableMetrics {
		this.pubMetrics.JobTryQps.Mark(1)
	}

	t1 := time.Now()
	realIp := getHttpRemoteIp(r)
	appid := r.Header.Get(HttpHeaderAppid)

	var due int64
	q := r.URL.Query()
	dueParam := q.Get("due") // due has higher priority than delay
	if dueParam != "" {
		d, err := strconv.ParseInt(dueParam, 10, 64)
		if err != nil {
			log.Error("+job[%s] %s(%s) due:%s %s", appid, r.RemoteAddr, realIp, dueParam, err)

			writeBadRequest(w, "invalid due param")
			return
		}

		due = d
	} else {
		delayParam := q.Get("delay") // in sec
		delay, err := strconv.ParseInt(delayParam, 10, 64)
		if err != nil {
			log.Error("+job[%s] %s(%s) delay:%s %s", appid, r.RemoteAddr, realIp, delayParam, err)

			writeBadRequest(w, "invalid delay param")
			return
		}

		due = t1.Unix() + delay
	}

	if due <= t1.Unix() {
		log.Error("+job[%s] %s(%s) due=%d before now?", appid, r.RemoteAddr, realIp, due)

		writeBadRequest(w, "invalid param")
		return
	}

	if Options.Ratelimit && !this.throttlePub.Pour(realIp, 1) {
		log.Warn("+job[%s] %s(%s) rate limit reached", appid, r.RemoteAddr, realIp)

		writeQuotaExceeded(w)
		return
	}

	topic := params.ByName(UrlParamTopic)
	ver := params.ByName(UrlParamVersion)
	if err := manager.Default.OwnTopic(appid, r.Header.Get(HttpHeaderPubkey), topic); err != nil {
		log.Warn("+job[%s] %s(%s) {topic:%s, ver:%s} %s", appid, r.RemoteAddr, realIp, topic, ver, err)

		writeAuthFailure(w, err)
		return
	}

	// get the raw POST message
	msgLen := int(r.ContentLength)
	switch {
	case msgLen == -1:
		log.Warn("+job[%s] %s(%s) {topic:%s, ver:%s} invalid content length: %d",
			appid, r.RemoteAddr, realIp, topic, ver, msgLen)

		writeBadRequest(w, "invalid content length")
		return

	case int64(msgLen) > Options.MaxJobSize:
		log.Warn("+job[%s] %s(%s) {topic:%s, ver:%s} too big content length: %d",
			appid, r.RemoteAddr, realIp, topic, ver, msgLen)
		writeBadRequest(w, ErrTooBigMessage.Error())
		return

	case msgLen < Options.MinPubSize:
		log.Warn("+job[%s] %s(%s) {topic:%s, ver:%s} too small content length: %d",
			appid, r.RemoteAddr, realIp, topic, ver, msgLen)
		writeBadRequest(w, ErrTooSmallMessage.Error())
		return
	}

	lbr := io.LimitReader(r.Body, Options.MaxJobSize+1)
	msg := mpool.NewMessage(msgLen)
	msg.Body = msg.Body[0:msgLen]
	if _, err := io.ReadAtLeast(lbr, msg.Body, msgLen); err != nil {
		msg.Free()

		log.Error("+job[%s] %s(%s) {topic:%s, ver:%s} %s",
			appid, r.RemoteAddr, realIp, topic, ver, err)
		writeBadRequest(w, ErrTooBigMessage.Error()) // TODO http.StatusRequestEntityTooLarge
		return
	}

	log.Debug("+job[%s] %s(%s) {topic:%s, ver:%s} due:%d/%ds",
		appid, r.RemoteAddr, realIp, topic, ver, due, due-t1.Unix())

//.........这里部分代码省略.........
开发者ID:funkygao，项目名称:gafka，代码行数:101，代码来源:handler_job.go


示例17: subWsHandler

//go:generate goannotation $GOFILE
// @rest GET /v1/ws/msgs/:appid/:topic/:ver?group=xx
func (this *subServer) subWsHandler(w http.ResponseWriter, r *http.Request, params httprouter.Params) {
	ws, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		log.Error("%s: %v", r.RemoteAddr, err)
		return
	}

	defer func() {
		ws.Close()

		this.gw.svrMetrics.ConcurrentSubWs.Dec(1)
		this.idleConnsWg.Done()
	}()

	var (
		topic       string
		ver         string
		myAppid     string
		hisAppid    string
		resetOffset string
		group       string
	)

	query := r.URL.Query()
	group = query.Get("group")
	resetOffset = query.Get("reset")
	limit, err := getHttpQueryInt(&query, "limit", 1)
	if err != nil {
		writeWsError(ws, err.Error())
		return
	}
	if !manager.Default.ValidateGroupName(r.Header, group) {
		log.Warn("consumer %s{topic:%s, ver:%s, group:%s, limit:%d} invalid group",
			r.RemoteAddr, topic, ver, group, limit)
		return
	}

	ver = params.ByName(UrlParamVersion)
	topic = params.ByName(UrlParamTopic)
	hisAppid = params.ByName(UrlParamAppid)
	myAppid = r.Header.Get(HttpHeaderAppid)
	realIp := getHttpRemoteIp(r)
	if err = manager.Default.AuthSub(myAppid, r.Header.Get(HttpHeaderSubkey),
		hisAppid, topic, group); err != nil {
		log.Error("consumer[%s] %s {hisapp:%s, topic:%s, ver:%s, group:%s, limit:%d}: %s",
			myAppid, r.RemoteAddr, hisAppid, topic, ver, group, limit, err)

		writeWsError(ws, "auth fail")
		return
	}

	log.Debug("sub[%s] %s: %+v", myAppid, r.RemoteAddr, params)

	rawTopic := manager.Default.KafkaTopic(hisAppid, topic, ver)
	cluster, found := manager.Default.LookupCluster(hisAppid)
	if !found {
		log.Error("cluster

鲜花
握手
雷人
路过
鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
More+
上一篇：
Golang jsconf.Conf类代码示例发布时间：2022-05-23
下一篇：
Golang gofmt.PrettySince函数代码示例发布时间：2022-05-23
热门推荐
More+
热门话题
More+
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服（服务时间 9:00～18:00）

在线QQ客服
地址：深圳市南山区西丽大学城创智工业园
电邮：jeky_zhao#qq.com
移动电话：139-2527-9053

Powered by 互联科技 X3.4© 2001-2213 极客世界.|Sitemap