作者 陈志颖

合并分支 'dev' 到 'test'

Dev



查看合并请求 !53
... ... @@ -3,11 +3,12 @@ package factory
import (
"github.com/linmadan/egglib-go/core/application"
"gitlab.fjmaimaimai.com/mmm-go/partnermg/pkg/infrastructure/pg"
"gitlab.fjmaimaimai.com/mmm-go/partnermg/pkg/infrastructure/pg/transaction"
pG "gitlab.fjmaimaimai.com/mmm-go/partnermg/pkg/infrastructure/pg/transaction"
)
func CreateTransactionContext(options map[string]interface{}) (application.TransactionContext, error) {
return &transaction.TransactionContext{
PgDd: pg.DB,
}, nil
//return &transaction.TransactionContext{
// PgDd: pg.DB,
//}, nil
return pG.NewPGTransactionContext(pg.DB), nil
}
... ...
... ... @@ -29,7 +29,9 @@ type CreateOrderFromBestshop struct {
DeliveryTime string `json:"deliveryTime"`
PartnerId int64 `json:"partnerId"`
OrderArea string `json:"orderArea"`
Goods []struct {
//小程序id
WxAppletId string `json:"wxAppletId"`
Goods []struct {
Id int64 `json:"id"`
//货品编号
Sn string `json:"sn"`
... ...
... ... @@ -37,7 +37,7 @@ func (s SyncOrderService) SyncOrderFromBestshop(cmd command.CreateOrderFromBests
return lib.ThrowError(lib.INTERNAL_SERVER_ERROR, err.Error())
}
defer func() {
transactionContext.RollbackTransaction()
_ = transactionContext.RollbackTransaction()
}()
//检查账号是否存在
... ... @@ -101,7 +101,7 @@ func (s SyncOrderService) CreateOrderFromBestshop(cmd command.CreateOrderFromBes
return lib.ThrowError(lib.INTERNAL_SERVER_ERROR, err.Error())
}
defer func() {
transactionContext.RollbackTransaction()
_ = transactionContext.RollbackTransaction()
}()
var (
orderBestshopRepository domain.OrderBestshopRepository
... ... @@ -117,6 +117,7 @@ func (s SyncOrderService) CreateOrderFromBestshop(cmd command.CreateOrderFromBes
}); err != nil {
return lib.ThrowError(lib.INTERNAL_SERVER_ERROR, err.Error())
}
// TODO 增加小程序id
order := domain.OrderBestShop{
OrderCode: cmd.OrderCode,
OrderTime: cmd.OrderTime,
... ... @@ -135,6 +136,7 @@ func (s SyncOrderService) CreateOrderFromBestshop(cmd command.CreateOrderFromBes
IsCopy: false,
CompanyId: cmd.CompanyId,
OrderArea: cmd.OrderArea,
WxAppletId: cmd.WxAppletId,
}
err = orderBestshopRepository.Add(&order)
if err != nil {
... ... @@ -232,11 +234,16 @@ func (s SyncOrderService) copyOrderBestshopToOrderBase(orderBestshop *domain.Ord
e := fmt.Sprintf("未找到指定的合伙人的公司(partner_id=%d,company_id=%d)数据,%s", orderBestshop.PartnerId, partnerData.CompanyId, err)
return lib.ThrowError(lib.INTERNAL_SERVER_ERROR, e)
}
// TODO 判断多小程序
for _, v := range companyData.Applets {
//BEST_SHOP_UNIONID string = "gh_18eb644002fb" //香米小程序原始id
//接收香米小程序的订单数据
if len(v.Id) > 0 {
//if len(v.Id) > 0 {
// canCopyOrder = true
//}
if v.Id == orderBestshop.WxAppletId {
canCopyOrder = true
break
}
}
if !canCopyOrder {
... ... @@ -329,7 +336,7 @@ func (s SyncOrderService) UpdateOrderFromBestshop(cmd command.CreateOrderFromBes
if err != nil {
return lib.ThrowError(lib.INTERNAL_SERVER_ERROR, "获取orderBestshop(order_code=%s)数据失败,err=%s", cmd.OrderCode, err.Error())
}
// TODO 增加小程序id
orderData.OrderCode = cmd.OrderCode
orderData.OrderTime = cmd.OrderTime
orderData.OrderState = cmd.OrderState
... ... @@ -344,6 +351,7 @@ func (s SyncOrderService) UpdateOrderFromBestshop(cmd command.CreateOrderFromBes
orderData.DeliveryState = cmd.DeliveryState
orderData.DeliveryTime = cmd.DeliveryTime
orderData.CompanyId = cmd.CompanyId
orderData.WxAppletId = cmd.WxAppletId
err = orderBestshopRepository.Edit(orderData)
if err != nil {
return lib.ThrowError(lib.INTERNAL_SERVER_ERROR, "编辑order_bestshop失败:"+err.Error())
... ...
... ... @@ -41,6 +41,8 @@ type OrderBestShop struct {
CompanyId int64 `json:"companyId"`
//订单区域
OrderArea string `json:"orderArea"`
// 微信小程序id
WxAppletId string `jsons:"wxAppletId"`
}
func (order OrderBestShop) CopyToOrderBase(o *OrderBase) {
... ...
... ... @@ -21,7 +21,7 @@ type OrderBestshop struct {
BuyerAddress string
//买家备注
BuyerRemark string
//
//买家id
BuyerId int64
//订单总数
OrderCount int
... ... @@ -31,9 +31,14 @@ type OrderBestshop struct {
DeliveryTime string
//创建时间
CreateTime time.Time
PartnerId int64
//合伙人id
PartnerId int64
//是否将数据同步到 order_base ,order_good
IsCopy bool `pg:",use_zero"`
IsCopy bool `pg:",use_zero"`
//公司id
CompanyId int64
//订单区域
OrderArea string
//微信小程序id
WxAppletId string
}
... ...
... ... @@ -44,6 +44,7 @@ func (respository OrderBestshopRepository) transformPgModelToDomainModel(orderMo
IsCopy: orderModel.IsCopy,
CompanyId: orderModel.CompanyId,
OrderArea: orderModel.OrderArea,
WxAppletId: orderModel.WxAppletId,
}, nil
}
... ... @@ -67,6 +68,7 @@ func (respository OrderBestshopRepository) Add(order *domain.OrderBestShop) erro
IsCopy: order.IsCopy,
CompanyId: order.CompanyId,
OrderArea: order.OrderArea,
WxAppletId: order.WxAppletId,
}
_, err := tx.Model(&m).Insert()
order.Id = m.Id
... ... @@ -94,6 +96,7 @@ func (respository OrderBestshopRepository) Edit(order *domain.OrderBestShop) err
IsCopy: order.IsCopy,
CompanyId: order.CompanyId,
OrderArea: order.OrderArea,
WxAppletId: order.WxAppletId,
}
_, err := tx.Model(&m).Where("id=?", order.Id).Update()
order.Id = m.Id
... ...
... ... @@ -13,14 +13,18 @@ import (
func DataFromXiangMi(message *sarama.ConsumerMessage) error {
logs.Info("Done Message claimed: timestamp = %v, topic = %s offset = %v value = %v \n",
message.Timestamp, message.Topic, message.Offset, string(message.Value))
var (
msgData DataFromMessage
err error
)
err = json.Unmarshal(message.Value, &msgData)
if err != nil {
return fmt.Errorf("[Consumer][SyncBestshopOrder] 解析kafka数据失败;%s", err)
}
// TODO 使用小程序id作为module
dataAction := msgData.Module + "/" + msgData.Action
switch dataAction {
case "xiangmi.order/ship":
... ... @@ -29,6 +33,13 @@ func DataFromXiangMi(message *sarama.ConsumerMessage) error {
e := fmt.Errorf("[Consumer][SyncBestshopOrder] %s", err)
return e
}
// TODO 统一操作小程序订单
case "wxapplet.order/ship":
err = syncBestshopOrder(msgData.Data)
if err != nil {
e := fmt.Errorf("[Consumer][SyncBestshopOrder] %s", err)
return e
}
default:
logs.Error("未找到执行动作:Module=%s,Action=%s", msgData.Module, msgData.Action)
}
... ...
/**
@author: stevechan
@date: 2021/3/11
@note:
**/
package configs
type MqConfig struct {
Topics []string `json:"topics"`
Servers []string `json:"servers"`
ConsumerId string `json:"consumerGroup"`
}
var Cfg = MqConfig{
Topics: []string{"xiangmi_project_dev"},
Servers: []string{"127.0.0.1:9092"},
ConsumerId: "test",
}
... ...
/**
@author: stevechan
@date: 2021/3/11
@note:
**/
package sync_order
import (
"encoding/json"
"fmt"
"github.com/Shopify/sarama"
"github.com/gavv/httpexpect"
"github.com/go-pg/pg"
pg2 "github.com/go-pg/pg/v10"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
pG "gitlab.fjmaimaimai.com/mmm-go/partnermg/pkg/infrastructure/pg"
"gitlab.fjmaimaimai.com/mmm-go/partnermg/test/integration/consumer/sync_order/configs"
"net/http"
"strconv"
"time"
)
var (
cfg *configs.MqConfig
producer sarama.SyncProducer
)
func produce(topic string, key string, content interface{}) error {
data, _ := json.Marshal(content)
msg := &sarama.ProducerMessage{
Topic: topic,
Key: sarama.StringEncoder(key),
Value: sarama.ByteEncoder(data),
Timestamp: time.Now(),
}
_, _, err2 := producer.SendMessage(msg)
if err2 != nil {
msg1 := fmt.Sprintf("Send Error topic: %v. key: %v. content: %v", topic, key, content)
fmt.Println(msg1)
return err2
}
fmt.Printf("Send OK topic:%s key:%s value:%s\n", topic, key, content)
return nil
}
var _ = Describe("同步小程序订单", func() {
var companyId int64
var partnerId int64
BeforeEach(func() {
fmt.Print("init kafka producer, it may take a few seconds to init the connection\n")
var err error
cfg = &configs.Cfg
mqConfig := sarama.NewConfig()
mqConfig.Producer.Return.Successes = true
mqConfig.Version = sarama.V0_10_2_1
if err = mqConfig.Validate(); err != nil {
msg := fmt.Sprintf("Kafka producer config invalidate. config: %v. err: %v", *cfg, err)
fmt.Println(msg)
panic(msg)
}
producer, err = sarama.NewSyncProducer(cfg.Servers, mqConfig)
if err != nil {
msg := fmt.Sprintf("Kafak producer create fail. err: %v", err)
fmt.Println(msg)
panic(msg)
}
// 新增公司
_, err2 := pG.DB.QueryOne(
pg.Scan(&companyId),
"INSERT INTO company (name, phone, logo, admin_company, status, create_at, update_at, delete_at, remarks, enable, abbreviation, applets) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) RETURNING id",
"测试公司", "13162672794", "http://suplus-file-dev.fjmaimaimai.com/upload/image/2020091709304822611.jpg", 431, Default, "2020-09-17 09:30:05.588579+08", "2020-09-17 09:30:05.588579+08", pg2.NullTime{}, "", "测试", 1, "公司", []map[string]interface{}{{"id": "gh_18eb644002fb", "url": "pages/index/index", "name": "海鲜干货啦啦啦啦啦", "imageUrl": ""}})
Expect(err2).NotTo(HaveOccurred())
// 新增合伙人
_, err3 := pG.DB.QueryOne(
pg.Scan(&partnerId),
"INSERT INTO partner_info (partner_name, account, password, status, partner_category, region_info, cooperate_time, create_at, update_at, salesman, partner_category_infos, company_id, remark) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) RETURNING id",
"test_partner", "19121619631", "7c4a8d09ca3762af61e59520943dc26494f8941b", 1, 1, map[string]interface{}{"regionName": "123"}, "2020-09-01 00:00:00+08", "2020-09-01 00:00:00+08", "2020-09-01 00:00:00+08", []map[string]interface{}{{"name": "123", "telephone": "13322223333"}}, []map[string]interface{}{{"id": 2, "code": "001"}}, 1, "测试合伙人")
Expect(err3).NotTo(HaveOccurred())
key := strconv.FormatInt(time.Now().UTC().UnixNano(), 10)
value := map[string]interface{}{
"module": "wxapplet.order",
"action": "ship",
"data": map[string]interface{}{
"orderCode": "1234567895",
"orderCount": 1,
"orderAmount": 2,
"orderTime": "2021-03-05 12:01:54",
"orderState": 1,
"deliveryState": 1,
"buyerName": "陈志颖",
"buyerPhone": "19121619631",
"buyerAddress": "福建省 福州市 仓山区",
"buyerRemark": "不辣",
"buyerId": 123,
"partnerId": partnerId,
"goods": []map[string]interface{}{
{
"sn": "wertyuf",
"id": 1,
"bn": "a;slkdjgaj",
"name": "地瓜",
"price": 2,
"nums": 1,
"amount": 2,
},
},
"companyId": companyId,
"deliveryTime": "2021-03-05 12:01:54",
"orderArea": "福建省 福州市 台江区",
"wxAppletId": "gh_18eb644002fb",
},
}
err4 := produce(cfg.Topics[0], key, value)
Expect(err4).NotTo(HaveOccurred())
})
Describe("提交小程序订单消息", func() {
Context("提交正确的小程序订单", func() {
It("返回正确的订单数据", func() {
httpExpect := httpexpect.New(GinkgoT(), server.URL)
httpExpect.GET("").
Expect().
Status(http.StatusOK).
JSON().
Object().
ContainsKey("code").ValueEqual("code", 0)
})
})
})
AfterEach(func() {
_, err1 := pG.DB.Exec("DELETE FROM order_bestshop WHERE true")
Expect(err1).NotTo(HaveOccurred())
_, err2 := pG.DB.Exec("DELETE FROM company WHERE true")
Expect(err2).NotTo(HaveOccurred())
_, err3 := pG.DB.Exec("DELETE FROM partner_info WHERE true")
Expect(err3).NotTo(HaveOccurred())
})
})
... ...
/**
@author: stevechan
@date: 2021/3/11
@note:
**/
package sync_order
import (
"github.com/astaxie/beego"
"net/http"
"net/http/httptest"
"testing"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
_ "gitlab.fjmaimaimai.com/mmm-go/partnermg/pkg/infrastructure/pg"
_ "gitlab.fjmaimaimai.com/mmm-go/partnermg/pkg/port/beego"
)
func TestColumnSetting(t *testing.T) {
RegisterFailHandler(Fail)
RunSpecs(t, "Kafka Consumer Sync Order Correlations Test Case Suite")
}
var handler http.Handler
var server *httptest.Server
// 初始化kafka
var _ = BeforeSuite(func() {
handler = beego.BeeApp.Handlers
server = httptest.NewServer(handler)
})
var _ = AfterSuite(func() {
server.Close()
})
... ...