Merge remote-tracking branch 'origin/dev'

@@ -14,7 +14,7 @@ require (
 	//github.com/opentracing/opentracing-go v1.1.1-0.20190913142402-a7454ce5950e
 	//github.com/opentracing/opentracing-go v1.1.1-0.20190913142402-a7454ce5950e
 	//github.com/openzipkin-contrib/zipkin-go-opentracing v0.4.5
 	//github.com/openzipkin-contrib/zipkin-go-opentracing v0.4.5
 	//github.com/openzipkin/zipkin-go v0.2.5
 	//github.com/openzipkin/zipkin-go v0.2.5
-	github.com/tiptok/gocomm v1.0.6
+	github.com/tiptok/gocomm v1.0.9
 )
 )
 
 
-//replace github.com/tiptok/gocomm v1.0.5 => F:\go\src\learn_project\gocomm
+//replace github.com/tiptok/gocomm v1.0.8 => F:\go\src\learn_project\gocomm

@@ -57,5 +57,8 @@ func main() {
 	//}
 	//}
 	log.Debug(constant.REDIS_HOST, constant.REDIS_PORT)
 	log.Debug(constant.REDIS_HOST, constant.REDIS_PORT)
 
 
+	//go message.RunConsumer()
+	//go message.RunPublish()
+
 	beego.Run()
 	beego.Run()
 }
 }

@@ -10,9 +10,9 @@ type CreateClientVersionRequest struct {
 	// 提交人
 	// 提交人
 	//Commiter int64 `json:"commiter,omitempty"`
 	//Commiter int64 `json:"commiter,omitempty"`
 	// 项目名称
 	// 项目名称
-	ProjectName string `json:"projectName,omitempty"`
+	ProjectName string `json:"projectName,omitempty" valid:"Required"`
 	// 版本号
 	// 版本号
-	Version string `json:"version,omitempty"`
+	Version string `json:"version,omitempty" valid:"Required"`
 	// 标题
 	// 标题
 	Title string `json:"title,omitempty"`
 	Title string `json:"title,omitempty"`
 	// 其他备注信息
 	// 其他备注信息

@@ -163,11 +163,15 @@ func (svr *ClientVersionService) ListClientVersion(header *protocol.RequestHeade
 	if total, clientVersion, err = ClientVersionRepository.Find(common.ObjectToMap(request)); err != nil {
 	if total, clientVersion, err = ClientVersionRepository.Find(common.ObjectToMap(request)); err != nil {
 		return
 		return
 	}
 	}
-	rsp = map[string]interface{}{
-		"totalRow":   total,
-		"pageNumber": (request.Offset + request.Limit) / request.Limit,
-		"lists":      utils.LoadCustomField(clientVersion, "Id", "ProjectName", "Version", "Title", "Remark", "CreateTime"),
+	rspData := map[string]interface{}{
+		"totalRow": total,
+		"lists":    utils.LoadCustomField(clientVersion, "Id", "ProjectName", "Version", "Title", "Remark", "ClientPackageInfo", "CreateTime"),
 	}
 	}
+	if request.Limit > 0 {
+		rspData["pageNumber"] = (request.Offset + request.Limit) / request.Limit
+	}
+	rsp = rspData
+
 	err = transactionContext.CommitTransaction()
 	err = transactionContext.CommitTransaction()
 	return
 	return
 }
 }

+package message
+
+import (
+	"github.com/tiptok/gocomm/pkg/broker"
+	"github.com/tiptok/gocomm/pkg/broker/local"
+	"github.com/tiptok/gocomm/pkg/broker/models"
+	"github.com/tiptok/gocomm/pkg/log"
+	"gitlab.fjmaimaimai.com/mmm-go/godevp/pkg/domain"
+	"gitlab.fjmaimaimai.com/mmm-go/godevp/pkg/infrastructure/pg"
+	"time"
+)
+
+func RunPublish() {
+	var (
+		err error
+	)
+
+	producer := broker.NewMessageProducer(
+		models.WithKafkaHost(KAFKA_HOSTS),
+		models.WithMessageProduceRepository(local.NewPgMessageProduceRepository(pg.DB, nil)),
+	)
+	for {
+		messages := []interface{}{
+			domain.Users{
+				Id:         20200129,
+				Name:       "user_2020",
+				Phone:      "18860183058",
+				Status:     1,
+				Roles:      []int64{7, 8},
+				CreateTime: time.Now(),
+				UpdateTime: time.Now(),
+			},
+		}
+		err = producer.Publish(TOPIC_BUSSINESS_ADMIN, messages, models.WithMessageProduceRepository(local.NewPgMessageProduceRepository(pg.DB, nil)))
+		if err != nil {
+			log.Error(err)
+		}
+		log.Debug("publish :", messages)
+		time.Sleep(time.Hour * 24)
+	}
+}

+package message
+
+import (
+	"github.com/tiptok/gocomm/common"
+	"github.com/tiptok/gocomm/pkg/broker/kafkax"
+	"github.com/tiptok/gocomm/pkg/broker/local"
+	"github.com/tiptok/gocomm/pkg/broker/models"
+	"github.com/tiptok/gocomm/pkg/log"
+	"gitlab.fjmaimaimai.com/mmm-go/godevp/pkg/domain"
+	"gitlab.fjmaimaimai.com/mmm-go/godevp/pkg/infrastructure/pg"
+)
+
+const (
+	KAFKA_HOSTS           = "106.52.15.41:9092" // "192.168.0.250:9092,192.168.0.251:9092,192.168.0.252:9092" //"106.52.15.41:9092" //
+	GroupId               = "godevp_test"       // "calf_basic_test"                                          //"godevp"
+	TOPIC_BUSSINESS_ADMIN = "mmm_business_test"
+)
+
+func RunConsumer() {
+	saramaConsumer := kafkax.NewSaramaConsumer(KAFKA_HOSTS, GroupId) //models.WithHandlerOriginalMessageFlag(true)   models.WithVersion("0.10.2.1")
+	saramaConsumer.WithTopicHandler(TOPIC_BUSSINESS_ADMIN, UserLoginHandler)
+	saramaConsumer.WithMessageReceiver(local.NewPgMessageReceiverRepository(pg.DB, nil)) // 持久化
+
+	err := saramaConsumer.StartConsume()
+	if err != nil {
+		log.Error(err)
+	}
+}
+
+func UserLoginHandler(message interface{}) error {
+	msg, ok := message.(*models.Message)
+	if ok {
+		var user = &domain.Users{}
+		common.JsonUnmarshal(string(msg.Value), user)
+		log.Info("消费消息:", msg.Id, msg.Topic, msg.Value)
+		log.Info("登录用户信息:", user.Id, user.Name)
+	}
+	return nil
+}