init

circlebuffer.go

@@ -0,0 +1,158 @@
+package starnet
+
+import (
+	"errors"
+	"fmt"
+	"io"
+	"time"
+)
+
+type CircleByteBuffer struct {
+	io.Reader
+	io.Writer
+	io.Closer
+	datas []byte
+
+	start   int
+	end     int
+	size    int
+	isClose bool
+	isEnd   bool
+}
+
+func NewCircleByteBuffer(len int) *CircleByteBuffer {
+	var e = new(CircleByteBuffer)
+	e.datas = make([]byte, len)
+	e.start = 0
+	e.end = 0
+	e.size = len
+	e.isClose = false
+	e.isEnd = false
+	return e
+}
+
+func (e *CircleByteBuffer) getLen() int {
+	if e.start == e.end {
+		return 0
+	} else if e.start < e.end {
+		return e.end - e.start
+	} else {
+		return e.start - e.end
+	}
+}
+func (e *CircleByteBuffer) getFree() int {
+	return e.size - e.getLen()
+}
+func (e *CircleByteBuffer) putByte(b byte) error {
+	if e.isClose {
+		return io.EOF
+	}
+	e.datas[e.end] = b
+	var pos = e.end + 1
+	for pos == e.start {
+		if e.isClose {
+			return io.EOF
+		}
+		time.Sleep(time.Microsecond)
+	}
+	if pos == e.size {
+		e.end = 0
+	} else {
+		e.end = pos
+	}
+	return nil
+}
+
+func (e *CircleByteBuffer) getByte() (byte, error) {
+	if e.isClose {
+		return 0, io.EOF
+	}
+	if e.isEnd && e.getLen() <= 0 {
+		return 0, io.EOF
+	}
+	if e.getLen() <= 0 {
+		return 0, errors.New("no datas")
+	}
+	var ret = e.datas[e.start]
+	e.start++
+	if e.start == e.size {
+		e.start = 0
+	}
+	return ret, nil
+}
+func (e *CircleByteBuffer) geti(i int) byte {
+	if i >= e.getLen() {
+		panic("out buffer")
+	}
+	var pos = e.start + i
+	if pos >= e.size {
+		pos -= e.size
+	}
+	return e.datas[pos]
+}
+
+/*func (e*CircleByteBuffer)puts(bts []byte){
+	for i:=0;i<len(bts);i++{
+		e.put(bts[i])
+	}
+}
+func (e*CircleByteBuffer)gets(bts []byte)int{
+	if bts==nil {return 0}
+	var ret=0
+	for i:=0;i<len(bts);i++{
+		if e.getLen()<=0{break}
+		bts[i]=e.get()
+		ret++
+	}
+	return ret
+}*/
+func (e *CircleByteBuffer) Close() error {
+	e.isClose = true
+	return nil
+}
+func (e *CircleByteBuffer) Read(bts []byte) (int, error) {
+	if e.isClose {
+		return 0, io.EOF
+	}
+	if bts == nil {
+		return 0, errors.New("bts is nil")
+	}
+	var ret = 0
+	for i := 0; i < len(bts); i++ {
+		b, err := e.getByte()
+		if err != nil {
+			if err == io.EOF {
+				return ret, err
+			}
+			return ret, nil
+		}
+		bts[i] = b
+		ret++
+	}
+	if e.isClose {
+		return ret, io.EOF
+	}
+	return ret, nil
+}
+func (e *CircleByteBuffer) Write(bts []byte) (int, error) {
+	if e.isClose {
+		return 0, io.EOF
+	}
+	if bts == nil {
+		e.isEnd = true
+		return 0, io.EOF
+	}
+	var ret = 0
+	for i := 0; i < len(bts); i++ {
+		err := e.putByte(bts[i])
+		if err != nil {
+			fmt.Println("Write bts err:", err)
+			return ret, err
+		}
+		ret++
+	}
+	if e.isClose {
+		return ret, io.EOF
+	}
+	return ret, nil
+}

curl.go

@@ -0,0 +1,125 @@
+package starnet
+
+import (
+	"bytes"
+	"crypto/rand"
+	"errors"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"net"
+	"net/http"
+	"net/url"
+	"time"
+)
+
+type Request struct {
+	TimeOut       int
+	DialTimeOut   int
+	Url           string
+	Method        string
+	RecvData      []byte
+	WriteRecvData bool
+	RecvIo        io.Writer
+	ReqHeader     http.Header
+	ReqCookies    []*http.Cookie
+	RespHeader    http.Header
+	RespCookies   []*http.Cookie
+	RespHttpCode  int
+	PostBuffer    *bytes.Buffer
+	CircleBuffer  *CircleByteBuffer
+	Proxy         string
+}
+
+func NewRequests(url string, postdata []byte, method string) Request {
+	return Request{
+		TimeOut:     30,
+		DialTimeOut: 15,
+		Url:         url,
+		PostBuffer:  bytes.NewBuffer(postdata),
+		Method:      method,
+	}
+}
+
+func (curl *Request) ResetReqHeader() {
+	curl.ReqHeader = make(http.Header)
+}
+
+func (curl *Request) ResetReqCookies() {
+	curl.ReqCookies = []*http.Cookie{}
+}
+
+func (curl *Request) AddSimpleCookie(key, value string) {
+	curl.ReqCookies = append(curl.ReqCookies, &http.Cookie{Name: key, Value: value, Path: "/"})
+}
+
+func randomBoundary() string {
+	var buf [30]byte
+	_, err := io.ReadFull(rand.Reader, buf[:])
+	if err != nil {
+		panic(err)
+	}
+	return fmt.Sprintf("%x", buf[:])
+}
+
+func Curl(curl Request) (resps Request, err error) {
+	var req *http.Request
+	if curl.Method == "" {
+		return Request{}, errors.New("Error Method Not Entered")
+	}
+	if curl.PostBuffer != nil {
+		req, err = http.NewRequest(curl.Method, curl.Url, curl.PostBuffer)
+	} else {
+		req, err = http.NewRequest(curl.Method, curl.Url, curl.CircleBuffer)
+	}
+	if err != nil {
+		return
+	}
+	req.Header = curl.ReqHeader
+	if len(curl.ReqCookies) != 0 {
+		for _, v := range curl.ReqCookies {
+			req.AddCookie(v)
+		}
+	}
+	transport := &http.Transport{
+		Dial: func(netw, addr string) (net.Conn, error) {
+			deadline := time.Now().Add(time.Duration(curl.TimeOut) * time.Second)
+			c, err := net.DialTimeout(netw, addr, time.Second*time.Duration(curl.DialTimeOut))
+			if err != nil {
+				return nil, err
+			}
+			if curl.TimeOut != 0 {
+				c.SetDeadline(deadline)
+			}
+			return c, nil
+		},
+	}
+	if curl.Proxy != "" {
+		purl, _ := url.Parse(curl.Proxy)
+		transport.Proxy = http.ProxyURL(purl)
+	}
+	client := &http.Client{
+		Transport: transport,
+	}
+	resp, err := client.Do(req)
+	if err != nil {
+		return
+	}
+	defer resp.Body.Close()
+	curl.PostBuffer = nil
+	curl.CircleBuffer = nil
+	curl.RespHttpCode = resp.StatusCode
+	curl.RespHeader = resp.Header
+	curl.RespCookies = resp.Cookies()
+	body, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return
+	}
+	if curl.WriteRecvData {
+		curl.RecvData = body
+	}
+	if curl.RecvIo != nil {
+		_, err = curl.RecvIo.Write(body)
+	}
+	return curl, err
+}

http.go

@@ -0,0 +1,338 @@
+package starnet
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"mime/multipart"
+	"net"
+	"net/http"
+	urls "net/url"
+	"os"
+	"path/filepath"
+	"time"
+)
+
+type StarCurl struct {
+	TimeOut      int
+	DialTimeOut  int
+	ReqHeader    http.Header
+	ReqCookies   []*http.Cookie
+	RespHeader   http.Header
+	RespCookies  []*http.Cookie
+	RespHttpCode int
+	PostBuffer   *bytes.Buffer
+	CircleBuffer *CircleByteBuffer
+	Proxy        string
+}
+
+func NewStarCurl() *StarCurl {
+	star := new(StarCurl)
+	star.ReqHeader = make(http.Header)
+	star.TimeOut = 60
+	star.DialTimeOut = 15
+	star.PostBuffer = nil
+	return star
+}
+
+func (this *StarCurl) ResetReqHeader() {
+	this.ReqHeader = make(http.Header)
+}
+
+func (this *StarCurl) ResetReqCookies() {
+	this.ReqCookies = []*http.Cookie{}
+}
+
+func (this *StarCurl) AddSimpleCookie(key, value string) {
+	this.ReqCookies = append(this.ReqCookies, &http.Cookie{Name: key, Value: value, Path: "/"})
+}
+
+func (this *StarCurl) CurlWithFile(url string, postdata map[string]string, formname, fpath, savepath string, tofile bool, shell func(float64)) (result []byte, err error) {
+	fpsrc, err := os.Open(fpath)
+	if err != nil {
+		return
+	}
+	defer fpsrc.Close()
+	boundary := randomBoundary()
+	boundarybytes := []byte("\r\n--" + boundary + "\r\n")
+	endbytes := []byte("\r\n--" + boundary + "--\r\n")
+	fpstat, _ := os.Stat(fpath)
+	filebig := float64(fpstat.Size())
+	sum, n := 0, 0
+	fpdst := NewCircleByteBuffer(1048576)
+	if postdata != nil {
+		for k, v := range postdata {
+			header := fmt.Sprintf("Content-Disposition: form-data; name=\"%s\";\r\nContent-Type: x-www-form-urlencoded \r\n\r\n", k)
+			fpdst.Write(boundarybytes)
+			fpdst.Write([]byte(header))
+			fpdst.Write([]byte(v))
+		}
+	}
+	header := fmt.Sprintf("Content-Disposition: form-data; name=\"%s\"; filename=\"%s\"\r\nContent-Type: application/octet-stream\r\n\r\n", formname, fpstat.Name())
+	fpdst.Write(boundarybytes)
+	fpdst.Write([]byte(header))
+	go func() {
+		for {
+			bufs := make([]byte, 393213)
+			n, err = fpsrc.Read(bufs)
+			if err != nil {
+				if err == io.EOF {
+					if n != 0 {
+						fpdst.Write(bufs[0:n])
+						go shell(float64(sum+n) / filebig * 100)
+					}
+					break
+				}
+				return
+			}
+			sum += n
+			go shell(float64(sum) / filebig * 100)
+			fpdst.Write(bufs[0:n])
+		}
+		fpdst.Write(endbytes)
+		fpdst.Write(nil)
+	}()
+	this.CircleBuffer = fpdst
+	this.ReqHeader.Set("Content-Type", "multipart/form-data;boundary="+boundary)
+	if tofile {
+		err = this.CurlDataToFile(url, []byte{}, "POST", savepath, shell)
+		this.ResetReqHeader()
+	} else {
+		result, err = this.Curl(url, []byte{}, "POST")
+	}
+	this.ResetReqHeader()
+	return
+}
+
+func (this *StarCurl) CurlWithFileByBytes(url string, postdata map[string]string, formname, fname string, data []byte, savepath string, tofile bool) (result []byte, err error) {
+	buf := &bytes.Buffer{}
+	bufwriter := multipart.NewWriter(buf)
+	if postdata != nil {
+		for k, v := range postdata {
+			bufwriter.WriteField(k, v)
+		}
+	}
+	fpdst, err := bufwriter.CreateFormFile(formname, fname)
+	if err != nil {
+		return
+	}
+	fpdst.Write(data)
+	this.PostBuffer = buf
+	this.ReqHeader.Set("Content-Type", "multipart/form-data;boundary="+bufwriter.Boundary())
+	bufwriter.Close()
+	if tofile {
+		err = this.CurlDataToFile(url, []byte{}, "POST", savepath, func(float64) {})
+		this.ResetReqHeader()
+	} else {
+		result, err = this.Curl(url, []byte{}, "POST")
+	}
+	this.ResetReqHeader()
+	return
+}
+
+func (this *StarCurl) CurlWithFileByMemory(url string, postdata map[string]string, formname, fpath, savepath string, tofile bool, shell func(float64)) (result []byte, err error) {
+	buf := &bytes.Buffer{}
+	bufwriter := multipart.NewWriter(buf)
+	if postdata != nil {
+		for k, v := range postdata {
+			bufwriter.WriteField(k, v)
+		}
+	}
+	fpdst, err := bufwriter.CreateFormFile(formname, filepath.Base(fpath))
+	if err != nil {
+		return
+	}
+	fpsrc, err := os.Open(fpath)
+	if err != nil {
+		return
+	}
+	defer fpsrc.Close()
+	fpstat, _ := os.Stat(fpath)
+	filebig := float64(fpstat.Size())
+	sum, n := 0, 0
+	for {
+		bufs := make([]byte, 393213)
+		n, err = fpsrc.Read(bufs)
+		if err != nil {
+			if err == io.EOF {
+				if n != 0 {
+					fpdst.Write(bufs[0:n])
+					go shell(float64(sum+n) / filebig * 100)
+				}
+				break
+			}
+			return
+		}
+		sum += n
+		go shell(float64(sum) / filebig * 100)
+		fpdst.Write(bufs[0:n])
+	}
+
+	this.PostBuffer = buf
+	this.ReqHeader.Set("Content-Type", "multipart/form-data;boundary="+bufwriter.Boundary())
+	bufwriter.Close()
+	if tofile {
+		err = this.CurlDataToFile(url, []byte{}, "POST", savepath, shell)
+		this.ResetReqHeader()
+	} else {
+		result, err = this.Curl(url, []byte{}, "POST")
+	}
+	this.ResetReqHeader()
+	return
+}
+
+func (this *StarCurl) CurlDataToFile(url string, postdata []byte, method, fpath string, shell func(float64)) (err error) {
+	var req *http.Request
+	if method == "" {
+		if len(postdata) != 0 {
+			method = "POST"
+		} else {
+			method = "GET"
+		}
+	}
+	if len(postdata) == 0 && this.PostBuffer == nil && this.CircleBuffer == nil {
+		req, err = http.NewRequest(method, url, nil)
+	} else if len(postdata) != 0 {
+		req, err = http.NewRequest(method, url, bytes.NewBuffer(postdata))
+	} else if this.PostBuffer != nil {
+		req, err = http.NewRequest(method, url, this.PostBuffer)
+	} else {
+		req, err = http.NewRequest(method, url, this.CircleBuffer)
+	}
+	if err != nil {
+		return
+	}
+	if (this.ReqHeader == nil) && method == "POST" {
+		this.ReqHeader.Set("Content-Type", "application/x-www-form-urlencoded")
+	}
+	req.Header = this.ReqHeader
+	if len(this.ReqCookies) != 0 {
+		for _, v := range this.ReqCookies {
+			req.AddCookie(v)
+		}
+	}
+	transport := &http.Transport{
+		Dial: func(netw, addr string) (net.Conn, error) {
+			deadline := time.Now().Add(time.Duration(this.TimeOut) * time.Second)
+			c, err := net.DialTimeout(netw, addr, time.Second*time.Duration(this.DialTimeOut))
+			if err != nil {
+				return nil, err
+			}
+			if this.TimeOut != 0 {
+				c.SetDeadline(deadline)
+			}
+			return c, nil
+		},
+	}
+	if this.Proxy != "" {
+		purl, _ := urls.Parse(this.Proxy)
+		transport.Proxy = http.ProxyURL(purl)
+	}
+	client := &http.Client{
+		Transport: transport,
+	}
+	resp, err := client.Do(req)
+	if err != nil {
+		return
+	}
+	defer resp.Body.Close()
+	this.PostBuffer = nil
+	this.CircleBuffer = nil
+	this.RespHttpCode = resp.StatusCode
+	this.RespHeader = resp.Header
+	this.RespCookies = resp.Cookies()
+	fpsrc, err := os.Create(fpath)
+	if err != nil {
+		return
+	}
+	defer fpsrc.Close()
+	filebig := float64(resp.ContentLength)
+	if filebig <= 0 {
+		filebig = 100
+	}
+	var n, sum int = 0, 0
+	for {
+		buf := make([]byte, 393213)
+		n, err = resp.Body.Read(buf)
+		if err != nil {
+			if err == io.EOF {
+				err = nil
+				if n != 0 {
+					fpsrc.Write(buf[0:n])
+				}
+				go shell(100.00)
+				break
+			}
+			return
+		}
+		sum += n
+		go shell(float64(sum) / filebig * 100.00)
+		fpsrc.Write(buf[0:n])
+	}
+	return
+}
+
+func (this *StarCurl) Curl(url string, postdata []byte, method string) (body []byte, err error) {
+	var req *http.Request
+	if method == "" {
+		if len(postdata) != 0 {
+			method = "POST"
+		} else {
+			method = "GET"
+		}
+	}
+	if len(postdata) == 0 && this.PostBuffer == nil && this.CircleBuffer == nil {
+		req, err = http.NewRequest(method, url, nil)
+	} else if len(postdata) != 0 {
+		req, err = http.NewRequest(method, url, bytes.NewBuffer(postdata))
+	} else if this.PostBuffer != nil {
+		req, err = http.NewRequest(method, url, this.PostBuffer)
+	} else {
+		req, err = http.NewRequest(method, url, this.CircleBuffer)
+	}
+	if err != nil {
+		return
+	}
+	if (this.ReqHeader == nil) && method == "POST" {
+		this.ReqHeader.Set("Content-Type", "application/x-www-form-urlencoded")
+	}
+	req.Header = this.ReqHeader
+	if len(this.ReqCookies) != 0 {
+		for _, v := range this.ReqCookies {
+			req.AddCookie(v)
+		}
+	}
+	transport := &http.Transport{
+		Dial: func(netw, addr string) (net.Conn, error) {
+			deadline := time.Now().Add(time.Duration(this.TimeOut) * time.Second)
+			c, err := net.DialTimeout(netw, addr, time.Second*time.Duration(this.DialTimeOut))
+			if err != nil {
+				return nil, err
+			}
+			if this.TimeOut != 0 {
+				c.SetDeadline(deadline)
+			}
+			return c, nil
+		},
+	}
+	if this.Proxy != "" {
+		purl, _ := urls.Parse(this.Proxy)
+		transport.Proxy = http.ProxyURL(purl)
+	}
+	client := &http.Client{
+		Transport: transport,
+	}
+	resp, err := client.Do(req)
+	if err != nil {
+		return
+	}
+	defer resp.Body.Close()
+	this.PostBuffer = nil
+	this.CircleBuffer = nil
+	this.RespHttpCode = resp.StatusCode
+	this.RespHeader = resp.Header
+	this.RespCookies = resp.Cookies()
+	body, err = ioutil.ReadAll(resp.Body)
+	return
+}

que.go

@@ -0,0 +1,291 @@
+package starnet
+
+import (
+	"bytes"
+	"context"
+	"encoding/binary"
+	"errors"
+	"sync"
+	"time"
+)
+
+// 识别头
+var header = []byte{11, 27, 19, 96, 12, 25, 02, 20}
+
+// MsgQueue 为基本的信息单位
+type MsgQueue struct {
+	ID   uint16
+	Msg  []byte
+	Conn interface{}
+}
+
+// StarQueue 为流数据中的消息队列分发
+type StarQueue struct {
+	Encode     bool
+	Reserve    uint16
+	Msgid      uint16
+	MsgPool    []MsgQueue
+	UnFinMsg   sync.Map
+	LastID     int //= -1
+	ctx        context.Context
+	cancel     context.CancelFunc
+	duration   time.Duration
+	EncodeFunc func([]byte) []byte
+	DecodeFunc func([]byte) []byte
+}
+
+// NewQueue 建立一个新消息队列
+func NewQueue() *StarQueue {
+	var que StarQueue
+	que.Encode = false
+	que.ctx, que.cancel = context.WithCancel(context.Background())
+	que.duration = 0
+	return &que
+}
+
+// Uint32ToByte 4位uint32转byte
+func Uint32ToByte(src uint32) []byte {
+	res := make([]byte, 4)
+	res[3] = uint8(src)
+	res[2] = uint8(src >> 8)
+	res[1] = uint8(src >> 16)
+	res[0] = uint8(src >> 24)
+	return res
+}
+
+// ByteToUint32 byte转4位uint32
+func ByteToUint32(src []byte) uint32 {
+	var res uint32
+	buffer := bytes.NewBuffer(src)
+	binary.Read(buffer, binary.BigEndian, &res)
+	return res
+}
+
+// Uint16ToByte 2位uint16转byte
+func Uint16ToByte(src uint16) []byte {
+	res := make([]byte, 2)
+	res[1] = uint8(src)
+	res[0] = uint8(src >> 8)
+	return res
+}
+
+// ByteToUint16 用于byte转uint16
+func ByteToUint16(src []byte) uint16 {
+	var res uint16
+	buffer := bytes.NewBuffer(src)
+	binary.Read(buffer, binary.BigEndian, &res)
+	return res
+}
+
+// BuildMessage 生成编码后的信息用于发送
+func (que *StarQueue) BuildMessage(src []byte) []byte {
+	var buff bytes.Buffer
+	que.Msgid++
+	if que.Encode {
+		src = que.EncodeFunc(src)
+	}
+	length := uint32(len(src))
+	buff.Write(header)
+	buff.Write(Uint32ToByte(length))
+	buff.Write(Uint16ToByte(que.Msgid))
+	buff.Write(src)
+	return buff.Bytes()
+}
+
+// BuildHeader 生成编码后的Header用于发送
+func (que *StarQueue) BuildHeader(length uint32) []byte {
+	var buff bytes.Buffer
+	que.Msgid++
+	buff.Write(header)
+	buff.Write(Uint32ToByte(length))
+	buff.Write(Uint16ToByte(que.Msgid))
+	return buff.Bytes()
+}
+
+type unFinMsg struct {
+	ID         uint16
+	LengthRecv uint32
+	// HeaderMsg 信息头，应当为14位：8位识别码+4位长度码+2位id
+	HeaderMsg []byte
+	RecvMsg   []byte
+}
+
+// ParseMessage 用于解析收到的msg信息
+func (que *StarQueue) ParseMessage(msg []byte, conn interface{}) error {
+	tmp, ok := que.UnFinMsg.Load(conn)
+	if ok { //存在未完成的信息
+		lastMsg := tmp.(*unFinMsg)
+		headerLen := len(lastMsg.HeaderMsg)
+		if headerLen < 14 { //未完成头标题
+			//传输的数据不能填充header头
+			if len(msg) < 14-headerLen {
+				//加入header头并退出
+				lastMsg.HeaderMsg = bytesMerge(lastMsg.HeaderMsg, msg)
+				que.UnFinMsg.Store(conn, lastMsg)
+				return nil
+			}
+			//获取14字节完整的header
+			header := msg[0 : 14-headerLen]
+			lastMsg.HeaderMsg = bytesMerge(lastMsg.HeaderMsg, header)
+			//检查收到的header是否为认证header
+			//若不是，丢弃并重新来过
+			if !checkHeader(lastMsg.HeaderMsg[0:8]) {
+				que.UnFinMsg.Delete(conn)
+				if len(msg) == 0 {
+					return nil
+				}
+				return que.ParseMessage(msg, conn)
+			}
+			//获得本数据包长度
+			lastMsg.LengthRecv = ByteToUint32(lastMsg.HeaderMsg[8:12])
+			//获得本数据包ID
+			lastMsg.ID = ByteToUint16(lastMsg.HeaderMsg[12:14])
+			//存入列表
+			que.UnFinMsg.Store(conn, lastMsg)
+			msg = msg[14-headerLen:]
+			if uint32(len(msg)) < lastMsg.LengthRecv {
+				lastMsg.RecvMsg = msg
+				que.UnFinMsg.Store(conn, lastMsg)
+				return nil
+			}
+			if uint32(len(msg)) >= lastMsg.LengthRecv {
+				lastMsg.RecvMsg = msg[0:lastMsg.LengthRecv]
+				if que.Encode {
+					lastMsg.RecvMsg = que.DecodeFunc(lastMsg.RecvMsg)
+				}
+				msg = msg[lastMsg.LengthRecv:]
+				stroeMsg := MsgQueue{
+					ID:   lastMsg.ID,
+					Msg:  lastMsg.RecvMsg,
+					Conn: conn,
+				}
+				que.MsgPool = append(que.MsgPool, stroeMsg)
+				que.UnFinMsg.Delete(conn)
+				return que.ParseMessage(msg, conn)
+			}
+		} else {
+			lastID := int(lastMsg.LengthRecv) - len(lastMsg.RecvMsg)
+			if lastID < 0 {
+				que.UnFinMsg.Delete(conn)
+				return que.ParseMessage(msg, conn)
+			}
+			if len(msg) >= lastID {
+				lastMsg.RecvMsg = bytesMerge(lastMsg.RecvMsg, msg[0:lastID])
+				if que.Encode {
+					lastMsg.RecvMsg = que.DecodeFunc(lastMsg.RecvMsg)
+				}
+				stroeMsg := MsgQueue{
+					ID:   lastMsg.ID,
+					Msg:  lastMsg.RecvMsg,
+					Conn: conn,
+				}
+				que.MsgPool = append(que.MsgPool, stroeMsg)
+				que.UnFinMsg.Delete(conn)
+				if len(msg) == lastID {
+					return nil
+				}
+				msg = msg[lastID:]
+				return que.ParseMessage(msg, conn)
+			}
+			lastMsg.RecvMsg = bytesMerge(lastMsg.RecvMsg, msg)
+			que.UnFinMsg.Store(conn, lastMsg)
+			return nil
+		}
+	}
+	if len(msg) == 0 {
+		return nil
+	}
+	var start int
+	if start = searchHeader(msg); start == -1 {
+		return errors.New("data format error")
+	}
+	msg = msg[start:]
+	lastMsg := unFinMsg{}
+	que.UnFinMsg.Store(conn, &lastMsg)
+	return que.ParseMessage(msg, conn)
+}
+
+func checkHeader(msg []byte) bool {
+	if len(msg) != 8 {
+		return false
+	}
+	for k, v := range msg {
+		if v != header[k] {
+			return false
+		}
+	}
+	return true
+}
+
+func searchHeader(msg []byte) int {
+	if len(msg) < 8 {
+		return 0
+	}
+	for k, v := range msg {
+		find := 0
+		if v == header[0] {
+			for k2, v2 := range header {
+				if msg[k+k2] == v2 {
+					find++
+				} else {
+					break
+				}
+			}
+			if find == 8 {
+				return k
+			}
+		}
+	}
+	return -1
+}
+
+func bytesMerge(src ...[]byte) []byte {
+	var buff bytes.Buffer
+	for _, v := range src {
+		buff.Write(v)
+	}
+	return buff.Bytes()
+}
+
+// Restore 获取收到的信息
+func (que *StarQueue) Restore(n int) ([]MsgQueue, error) {
+	var res []MsgQueue
+	dura := time.Duration(0)
+	for len(que.MsgPool) < n {
+		select {
+		case <-que.ctx.Done():
+			return res, errors.New("Stoped By External Function Call")
+		default:
+			time.Sleep(time.Millisecond * 20)
+			dura = time.Millisecond*20 + dura
+			if que.duration != 0 && dura > que.duration {
+				return res, errors.New("Time Exceed")
+			}
+		}
+	}
+	if len(que.MsgPool) < n {
+		return res, errors.New("Result Not Enough")
+	}
+	res = que.MsgPool[0:n]
+	que.MsgPool = que.MsgPool[n:]
+	return res, nil
+}
+
+// RestoreOne 获取收到的一个信息
+func (que *StarQueue) RestoreOne() (MsgQueue, error) {
+	data, err := que.Restore(1)
+	if len(data) == 1 {
+		return data[0], err
+	}
+	return MsgQueue{}, err
+}
+
+// Stop 立即停止Restore
+func (que *StarQueue) Stop() {
+	que.cancel()
+}
+
+// RestoreDuration Restore最大超时时间
+func (que *StarQueue) RestoreDuration(tm time.Duration) {
+	que.duration = tm
+}