feat:xiugai

oss/grpc.go

@@ -1,6 +1,7 @@
 package oss
 
 import (
+	"fmt"
 	"jygit.jydev.jianyu360.cn/BaseService/ossClient/constant"
 	"jygit.jydev.jianyu360.cn/BaseService/ossClient/entity"
 	"jygit.jydev.jianyu360.cn/BaseService/ossClient/pb"
@@ -8,8 +9,7 @@ import (
 )
 
 type myWriter struct {
-	resp     pb.Service_DownloadServer
-	errorMsg string
+	resp pb.Service_DownloadServer
 }
 
 func (m *myWriter) Write(b []byte) (n int, err error) {
@@ -24,20 +24,19 @@ type Grpc struct {
 
 func (g *Grpc) Download(req *pb.DownloadRequest, resp pb.Service_DownloadServer) error {
 	if err := checkArgs(&entity.Args{BucketID: req.BucketID, ObjectName: req.ObjectName}); err != nil {
-		log.Println(constant.UploadFail, err)
+		log.Println("grpc方式", fmt.Sprintf(constant.DownloadFail, err))
 		return err
 	}
-	log.Println("grpc方式下载文件", req.BucketID, req.ObjectName)
-	return Download(&myWriter{resp: resp, errorMsg: constant.DownloadSuccess}, 0, req.BucketID, req.ObjectName)
+	log.Println("grpc方式下载", req.BucketID, req.ObjectName)
+	return Download(&myWriter{resp: resp}, 0, req.BucketID, req.ObjectName)
 }
 
 func (g *Grpc) GetBidDetail(req *pb.DownloadRequest, resp pb.Service_DownloadServer) error {
 	if err := checkArgs(&entity.Args{BucketID: req.BucketID, ObjectName: req.ObjectName}); err != nil {
-		log.Println(constant.GetBidDetailFail, err)
+		log.Println("grpc方式获取正文", err)
 		return err
 	}
 	log.Println("grpc方式获取正文", req.BucketID, req.ObjectName)
-	req.ObjectName += ".txt"
-	GetBidDetail(&myWriter{resp: resp, errorMsg: constant.GetBidDetailSuccess}, req.BucketID, req.ObjectName)
+	GetBidDetail(&myWriter{resp: resp}, req.BucketID, req.ObjectName)
 	return nil
 }

oss/http.go

@@ -137,7 +137,6 @@ func BidDetailHandler(w http.ResponseWriter, r *http.Request) {
 		http.Error(w, api.Error_msg_1002+"object_name", http.StatusInternalServerError)
 		return
 	}
-	objectName += ".txt"
 	log.Println(GetIp(r), "根据标讯id获取正文", bucketID, objectName)
 	GetBidDetail(w, bucketID, objectName)
 }

oss/oss.go

@@ -28,11 +28,11 @@ var (
 	DownLoadPool         = make(chan bool, g.Config().MustGet(gctx.New(), "downLoadPoolSize").Int())
 	GetDetailFromEsPool  = make(chan bool, g.Config().MustGet(gctx.New(), "getDetailFromEsPoolSize").Int())
 	GetDetailFromMgoPool = make(chan bool, g.Config().MustGet(gctx.New(), "getDetailFromMgoPoolSize").Int())
-	getDetail            = map[string]func(w io.Writer, bucketID, objectName string){
-		"oss": func(w io.Writer, bucketID, objectName string) {
-			Download(w, 0, bucketID, objectName)
+	getDetail            = map[string]func(w io.Writer, bucketID, objectName string) error{
+		"oss": func(w io.Writer, bucketID, objectName string) error {
+			return Download(w, 0, bucketID, objectName+".txt")
 		},
-		"es": func(w io.Writer, bucketID, objectName string) {
+		"es": func(w io.Writer, bucketID, objectName string) error {
 			GetDetailFromEsPool <- true
 			defer func() {
 				<-GetDetailFromEsPool
@@ -41,19 +41,23 @@ var (
 			list := es.VarEs.Get(indexName, indexName, fmt.Sprintf(`{"query":{"bool":{"filter":{"term":{"_id":"%s"}}}},"_source":["detail"]}`, objectName))
 			if list != nil && len(*list) > 0 {
 				detail, _ := (*list)[0]["detail"].(string)
-				io.Copy(w, bytes.NewReader([]byte(detail)))
+				_, err := io.Copy(w, bytes.NewReader([]byte(detail)))
+				return err
 			}
+			return errors.New("not find from es")
 		},
-		"mgo": func(w io.Writer, bucketID, objectName string) {
+		"mgo": func(w io.Writer, bucketID, objectName string) error {
 			GetDetailFromMgoPool <- true
 			defer func() {
 				<-GetDetailFromMgoPool
 			}()
-			data, _ := config.Mgo.FindOneByField(g.Config().MustGet(gctx.New(), "mongodb.collection").String(), objectName, `{"detail":1}`)
+			data, _ := config.Mgo.FindById(g.Config().MustGet(gctx.New(), "mongodb.collection").String(), objectName, `{"detail":1}`)
 			if data != nil && len(*data) > 0 {
 				detail, _ := (*data)["detail"].(string)
-				io.Copy(w, bytes.NewReader([]byte(detail)))
+				_, err := io.Copy(w, bytes.NewReader([]byte(detail)))
+				return err
 			}
+			return errors.New("not find from mgo")
 		},
 	}
 )
@@ -67,9 +71,10 @@ func GetBucket(bucketID string) (config.BucketInfo, error) {
 			return *v, nil
 		}
 	}
-	log.Println("Bucket %s not found in cache, querying database", bucketID)
+	err := fmt.Errorf("bucket id %s not found", bucketID)
+	log.Println(err)
 	// 模拟数据库查询：实际业务中这里应查询数据库
-	return config.BucketInfo{}, fmt.Errorf("bucket id %s not found", bucketID)
+	return config.BucketInfo{}, err
 }
 
 // bucketPool 缓存 bucket 对象，避免重复创建
@@ -215,7 +220,9 @@ func GetBidDetail(w io.Writer, bucketID, objectName string) {
 	atomic.AddInt64(&util.GetDetailCounter, 1)
 	defer atomic.AddInt64(&util.GetDetailCounter, -1)
 	for _, v := range g.Config().MustGet(gctx.New(), "getDetailOrder").Strings() {
-		getDetail[v](w, bucketID, objectName)
+		if err := getDetail[v](w, bucketID, objectName); err == nil {
+			break
+		}
 	}
 }
 

oss/rpc.go

@@ -3,6 +3,7 @@ package oss
 import (
 	"bytes"
 	"errors"
+	"fmt"
 	"io/ioutil"
 	"log"
 	"strings"
@@ -29,17 +30,17 @@ type OSSService struct {
  */
 func (s *OSSService) Upload(args *entity.UploadArgs, reply *api.Result) error {
 	if err := checkArgs(&entity.Args{BucketID: args.BucketID, ObjectName: args.ObjectName}); err != nil {
-		log.Println(err.Error())
+		log.Println("rpc方式", fmt.Sprintf(constant.UploadUrl, err))
 		return err
 	} else if args.Stream == nil || len(args.Stream) == 0 {
 		err := errors.New(api.Error_msg_1002 + "Stream")
-		log.Println(err.Error())
+		log.Println("rpc方式", fmt.Sprintf(constant.UploadUrl, err))
 		return err
 	}
 	log.Println("rpc方式上传文件", args.BucketID, args.ObjectName, args.Gzip, len(args.Stream))
 	err := Upload(args.BucketID, args.ObjectName, ioutil.NopCloser(bytes.NewReader(args.Stream)), args.Gzip)
 	if err != nil {
-		log.Printf(constant.UploadFail, err)
+		log.Println(args.BucketID, args.ObjectName, "rpc方式", fmt.Sprintf(constant.UploadUrl, err))
 		return err
 	}
 	reply.Error_msg = constant.UploadSuccess
@@ -57,13 +58,13 @@ func (s *OSSService) Upload(args *entity.UploadArgs, reply *api.Result) error {
  */
 func (s *OSSService) Download(args *entity.Args, reply *api.Result) error {
 	if err := checkArgs(args); err != nil {
-		log.Println(constant.UploadFail, err)
+		log.Println("rpc方式", fmt.Sprintf(constant.DownloadFail, err))
 		return err
 	}
-	log.Println("rpc方式下载文件", args.BucketID, args.ObjectName)
+	log.Println("rpc方式下载", args.BucketID, args.ObjectName)
 	b := &bytes.Buffer{}
 	if err := Download(b, 1, args.BucketID, args.ObjectName); err != nil {
-		log.Println(constant.DownloadFail, err)
+		log.Println(args.BucketID, args.ObjectName, "rpc方式", fmt.Sprintf(constant.DownloadFail, err))
 		return err
 	}
 	reply.Error_msg = constant.DownloadSuccess
@@ -82,13 +83,13 @@ func (s *OSSService) Download(args *entity.Args, reply *api.Result) error {
  */
 func (s *OSSService) Delete(args *entity.Args, reply *api.Result) error {
 	if err := checkArgs(args); err != nil {
-		log.Println(constant.UploadFail, err)
+		log.Println("rpc方式", fmt.Sprintf(constant.DeleteFail, err))
 		return err
 	}
 	log.Println("rpc方式删除文件", args.BucketID, args.ObjectName)
 	err := Delete(args.BucketID, args.ObjectName)
 	if err != nil {
-		log.Println(constant.DeleteFail, err)
+		log.Println(args.BucketID, args.ObjectName, "rpc方式", fmt.Sprintf(constant.DeleteFail, err))
 		return err
 	}
 	reply.Error_msg = constant.DeleteSuccess
@@ -107,11 +108,10 @@ func (s *OSSService) Delete(args *entity.Args, reply *api.Result) error {
  */
 func (s *OSSService) GetBidDetail(args *entity.Args, reply *api.Result) error {
 	if err := checkArgs(args); err != nil {
-		log.Println(constant.GetBidDetailFail, err)
+		log.Println("rpc方式", fmt.Sprintf(constant.GetBidDetailFail, err))
 		return err
 	}
 	log.Println("rpc方式获取正文", args.BucketID, args.ObjectName)
-	args.ObjectName += ".txt"
 	data := &strings.Builder{}
 	GetBidDetail(data, args.BucketID, args.ObjectName)
 	reply.Data = data.String()