Add Makefile

Signed-off-by: cai.zhang <[email protected]>

.github/workflows/main.yaml

@@ -11,6 +11,7 @@ on:
       - 'cmd/**'
       - '.github/workflows/main.yaml'
       - docker-compose.yml
+      - Makefile
       - '!**.md'
       - '!**_test.go'
   pull_request:
@@ -21,6 +22,7 @@ on:
       - 'cmd/**'
       - '.github/workflows/main.yaml'
       - docker-compose.yml
+      - Makefile
       - '!**.md'
       - '!**_test.go'
 

.golangci.yml

@@ -0,0 +1,32 @@
+run:
+  skip-dirs:
+    - build
+    - configs
+    - deployments
+    - docs
+    - scripts
+    - internal/core
+
+linters-settings:
+  golint:
+    min-confidence: 0
+
+  misspell:
+    locale: US
+
+linters:
+  disable-all: false
+  enable:
+    - typecheck
+    - goimports
+    - misspell
+    - govet
+    - golint
+    - ineffassign
+    - gosimple
+    - deadcode
+    - structcheck
+
+service:
+  golangci-lint-version: 1.27.0 # use the fixed version to not introduce new linters unexpectedly
+

Makefile

@@ -22,7 +22,7 @@ all: build-cpp build-go
 get-check-deps:
 	@mkdir -p ${GOPATH}/bin
 	@which golangci-lint 1>/dev/null || (echo "Installing golangci-lint" && curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(GOPATH)/bin v1.27.0)
-#	@which ruleguard 1>/dev/null || (echo "Installing ruleguard" && GO111MODULE=off $(GO) get github.com/quasilyte/go-ruleguard/...)
+	@which ruleguard 1>/dev/null || (echo "Installing ruleguard" && GO111MODULE=off $(GO) get github.com/quasilyte/go-ruleguard/...)
 
 get-build-deps:
 	@(env bash $(PWD)/scripts/install_deps.sh)
@@ -36,9 +36,13 @@ fmt:
 lint:
 	@echo "Running $@ check"
 	@GO111MODULE=on ${GOPATH}/bin/golangci-lint cache clean
-#	@GO111MODULE=on ${GOPATH}/bin/golangci-lint run --timeout=5m --config ./.golangci.yml
+	@GO111MODULE=on ${GOPATH}/bin/golangci-lint run --timeout=1m --config ./.golangci.yml || true
 
-verifiers: get-check-deps fmt lint
+ruleguard:
+	@echo "Running $@ check"
+	@${GOPATH}/bin/ruleguard -rules ruleguard.rule.go ./... || true
+
+verifiers: get-check-deps fmt lint ruleguard
 
 # Builds various components locally.
 build-go: verifiers
@@ -65,7 +69,7 @@ test-go: verifiers build-go
 	@(env bash $(PWD)/scripts/run_go_unittest.sh)
 
 test-cpp: build-cpp-with-unittest
-	@echo "Running cpp unittest..."
+	@echo "Running cpp unittests..."
 	@(env bash $(PWD)/scripts/run_cpp_unittest.sh)
 
 #TODO: build each component to docker

internal/msgstream/msgstream.go

@@ -2,7 +2,6 @@ package msgstream
 
 import (
 	"context"
-	"fmt"
 	"github.com/apache/pulsar-client-go/pulsar"
 	commonPb "github.com/zilliztech/milvus-distributed/internal/proto/commonpb"
 	"log"
@@ -142,7 +141,27 @@ func (ms *PulsarMsgStream) Produce(msgPack *MsgPack) commonPb.Status {
 			); err != nil {
 				log.Printf("post into pulsar filed, error = %v", err)
 			}
-			fmt.Println("send a msg")
+		}
+	}
+
+	return commonPb.Status{ErrorCode: commonPb.ErrorCode_SUCCESS}
+}
+
+func (ms *PulsarMsgStream) BroadCast(msgPack *MsgPack) commonPb.Status {
+	producerLen := len(ms.producers)
+	for _, v := range msgPack.Msgs {
+		mb, status := (*ms.msgMarshaler).Marshal(v)
+		if status.ErrorCode != commonPb.ErrorCode_SUCCESS {
+			log.Printf("Marshal ManipulationReqMsg failed, error ")
+			continue
+		}
+		for i := 0; i < producerLen; i++ {
+			if _, err := (*ms.producers[i]).Send(
+				context.Background(),
+				&pulsar.ProducerMessage{Payload: mb},
+			); err != nil {
+				log.Printf("post into pulsar filed, error = %v", err)
+			}
 		}
 	}
 

internal/msgstream/msgstream_test.go

@@ -94,6 +94,22 @@ func getTsMsg(msgType MsgType, reqId int64, hashValue int32) *TsMsg {
 			TimeSyncMsg: timeSyncResult,
 		}
 		tsMsg = timeSyncMsg
+	case kTimeTick:
+		insertRequest := internalPb.InsertRequest{
+			ReqType:        internalPb.ReqType_kTimeTick,
+			ReqId:          reqId,
+			CollectionName: "Collection",
+			PartitionTag:   "Partition",
+			SegmentId:      1,
+			ChannelId:      1,
+			ProxyId:        1,
+			Timestamps:     []uint64{1},
+		}
+		insertMsg := InsertTask{
+			HashValues: []int32{hashValue},
+			InsertRequest: insertRequest,
+		}
+		tsMsg = insertMsg
 	}
 	return &tsMsg
 }
@@ -211,3 +227,17 @@ func TestStream_TimeSync(t *testing.T) {
 	//run stream
 	initStream(pulsarAddress, producerChannels, consumerChannels, consumerSubName, &msgPack, kTimeSync, kTimeSync)
 }
+
+func TestStream_BroadCast(t *testing.T) {
+	pulsarAddress := "pulsar://localhost:6650"
+	producerChannels := []string{"insert"}
+	consumerChannels := []string{"insert"}
+	consumerSubName := "subInsert"
+
+	msgPack := MsgPack{}
+	msgPack.Msgs = append(msgPack.Msgs, getTsMsg(kTimeTick, 0, 0))
+	msgPack.Msgs = append(msgPack.Msgs, getTsMsg(kTimeTick, 3, 3))
+
+	//run stream
+	initStream(pulsarAddress, producerChannels, consumerChannels, consumerSubName, &msgPack, kInsert, kInsert)
+}