代码拉取完成，页面将自动刷新
同步操作将从 andeyalee/erpc 强制同步，此操作会覆盖自 Fork 仓库以来所做的任何修改，且无法恢复！！！
确定后同步将在后台操作，完成时将刷新页面，请耐心等待。
Teleport是一个通用、高效、灵活的TCP Socket框架。
可用于Peer-Peer对等通信、RPC、长连接网关、微服务、推送服务，游戏服务等领域。
官方QQ群：Go-Web 编程 42730308 
AB Testing 1: [Mac 4CPU 8GB] [single-process single-conn] teleport: QPS 41358
AB Testing 2: [Mac 4CPU 8GB] [single-process single-conn] teleport/socket: QPS 55419
| 版本 | 状态 | 分支 |
|---|---|---|
| v2 | release | master |
| v1 | release | v1 |
go get -u github.com/henrylee2cn/teleport
Header和Body两部分Header和Body编码类型，例如JSON Protobuf string
Body支持gzip压缩Header包含状态码及其描述文本Peer -> Connection -> Socket -> Session -> Context
每个数据包的内容如下:
type Packet struct {
// HeaderCodec header codec string
HeaderCodec string
// BodyCodec body codec string
BodyCodec string
// header content
Header *Header `json:"header"`
// body content
Body interface{} `json:"body"`
// header length
HeaderLength int64 `json:"header_length"`
// body length
BodyLength int64 `json:"body_length"`
// packet size
Size int64 `json:"size"`
}
其中头部内容为:
type Header struct {
// Packet id
Id string
// Service type
Type int32
// Service URI
Uri string
// Body gzip level [-2,9]
Gzip int32
// As reply, it indicates the service status code
StatusCode int32
// As reply, it indicates the service status text
Status string
}
默认的通信协议：
HeaderLength | HeaderCodecId | Header | BodyLength | BodyCodecId | Body
注意：
HeaderLength: uint32, 4 bytes, big endianHeaderCodecId: uint8, 1 byteHeader: header bytesBodyLength: uint32, 4 bytes, big endian
Body为空且不指明BodyCodecId
Body为空但是指明BodyCodecId
BodyCodecId: uint8, 1 byteBody: body bytes你可以通过实现接口的方法定制自己的通信协议：
// Protocol socket communication protocol
type Protocol interface {
// WritePacket writes header and body to the connection.
WritePacket(
packet *Packet,
destWriter *utils.BufioWriter,
tmpCodecWriterGetter func(codecName string) (*TmpCodecWriter, error),
isActiveClosed func() bool,
) error
// ReadPacket reads header and body from the connection.
ReadPacket(
packet *Packet,
bodyAdapter func() interface{},
srcReader *utils.BufioReader,
codecReaderGetter func(codecId byte) (*CodecReader, error),
isActiveClosed func() bool,
checkReadLimit func(int64) error,
) error
}
接着，你可以使用以下任意方式指定自己的通信协议：
func SetDefaultProtocol(socket.Protocol)
func (*Peer) ServeConn(conn net.Conn, protocol ...socket.Protocol) Session
func (*Peer) DialContext(ctx context.Context, addr string, protocol ...socket.Protocol) (Session, error)
func (*Peer) Dial(addr string, protocol ...socket.Protocol) (Session, error)
func (*Peer) Listen(protocol ...socket.Protocol) error
var cfg = &tp.PeerConfig{
DefaultReadTimeout: time.Minute * 3,
DefaultWriteTimeout: time.Minute * 3,
TlsCertFile: "",
TlsKeyFile: "",
SlowCometDuration: time.Millisecond * 500,
DefaultHeaderCodec: "protobuf",
DefaultBodyCodec: "json",
DefaultBodyGzipLevel: 5,
PrintBody: true,
DefaultDialTimeout: time.Second * 10,
ListenAddrs: []string{
"0.0.0.0:9090",
},
}
var peer = tp.NewPeer(cfg)
// It can be used as a server
peer.Listen()
// It can also be used as a client at the same time
var sess, err = peer.Dial("127.0.0.1:8080")
if err != nil {
tp.Panicf("%v", err)
}
// Home controller
type Home struct {
tp.PullCtx
}
// Test handler
func (h *Home) Test(args *[2]int) (int, tp.Xerror) {
a := (*args)[0]
b := (*args)[1]
return a + b, nil
}
// Msg controller
type Msg struct {
tp.PushCtx
}
// Test handler
func (m *Msg) Test(args *map[string]interface{}) {
tp.Infof("receive push(%s):\nargs: %#v\nquery: %#v\n", m.Ip(), args, m.Query())
}
func UnknownPullHandle(ctx tp.UnknownPullCtx, body *[]byte) (interface{}, tp.Xerror) {
var v interface{}
codecName, err := ctx.Unmarshal(*body, &v, true)
if err != nil {
return nil, tp.NewXerror(0, err.Error())
}
tp.Infof("receive unknown pull:\n codec: %s\n content: %#v", codecName, v)
return "this is reply string for unknown pull", nil
}
func UnknownPushHandle(ctx tp.UnknownPushCtx, body *[]byte) {
var v interface{}
codecName, err := ctx.Unmarshal(*body, &v, true)
if err != nil {
tp.Errorf("%v", err)
} else {
tp.Infof("receive unknown push:\n codec: %s\n content: %#v", codecName, v)
}
}
// AliasPlugin can be used to set aliases for pull or push services
type AliasPlugin struct {
Aliases map[string]string
}
// NewAliasPlugin creates a new NewAliasPlugin
func NewAliasPlugin() *AliasPlugin {
return &AliasPlugin{Aliases: make(map[string]string)}
}
// Alias sets a alias for the uri.
// For example Alias("/arith/mul", "/mul")
func (p *AliasPlugin) Alias(alias string, uri string) {
p.Aliases[alias] = uri
}
// Name return name of this plugin.
func (p *AliasPlugin) Name() string {
return "AliasPlugin"
}
// PostReadPullHeader converts the alias of this service.
func (p *AliasPlugin) PostReadPullHeader(ctx tp.ReadCtx) tp.Xerror {
var u = ctx.Input().Header.Uri
if p.Aliases != nil {
if a = p.Aliases[u]; a != "" {
ctx.Input().Header.Uri = a
}
}
return nil
}
aliasesPlugin := NewAliasPlugin()
aliasesPlugin.Alias("/alias", "/origin")
{
pullGroup := peer.PullRouter.Group("pull", aliasesPlugin)
pullGroup.Reg(new(Home))
peer.PullRouter.SetUnknown(UnknownPullHandle)
}
{
pushGroup := peer.PushRouter.Group("push")
pushGroup.Reg(new(Msg), aliasesPlugin)
peer.PushRouter.SetUnknown(UnknownPushHandle)
}
package main
import (
"time"
tp "github.com/henrylee2cn/teleport"
)
func main() {
go tp.GraceSignal()
tp.SetShutdown(time.Second*20, nil, nil)
var cfg = &tp.PeerConfig{
DefaultReadTimeout: time.Minute * 3,
DefaultWriteTimeout: time.Minute * 3,
TlsCertFile: "",
TlsKeyFile: "",
SlowCometDuration: time.Millisecond * 500,
DefaultHeaderCodec: "protobuf",
DefaultBodyCodec: "json",
DefaultBodyGzipLevel: 5,
PrintBody: true,
ListenAddrs: []string{
"0.0.0.0:9090",
"0.0.0.0:9091",
},
}
var peer = tp.NewPeer(cfg)
{
group := peer.PullRouter.Group("group")
group.Reg(new(Home))
}
peer.PullRouter.SetUnknown(UnknownPullHandle)
peer.Listen()
}
// Home controller
type Home struct {
tp.PullCtx
}
// Test handler
func (h *Home) Test(args *map[string]interface{}) (map[string]interface{}, tp.Xerror) {
h.Session().Push("/push/test?tag=from home-test", map[string]interface{}{
"your_id": h.Query().Get("peer_id"),
"a": 1,
})
return map[string]interface{}{
"your_args": *args,
"server_time": time.Now(),
}, nil
}
func UnknownPullHandle(ctx tp.UnknownPullCtx, body *[]byte) (interface{}, tp.Xerror) {
var v interface{}
codecName, err := ctx.Unmarshal(*body, &v, true)
if err != nil {
return nil, tp.NewXerror(0, err.Error())
}
tp.Debugf("unmarshal body: codec: %s, content: %#v", codecName, v)
return []string{"a", "aa", "aaa"}, nil
}
package main
import (
"time"
tp "github.com/henrylee2cn/teleport"
)
func main() {
go tp.GraceSignal()
tp.SetShutdown(time.Second*20, nil, nil)
var cfg = &tp.PeerConfig{
DefaultReadTimeout: time.Minute * 3,
DefaultWriteTimeout: time.Minute * 3,
TlsCertFile: "",
TlsKeyFile: "",
SlowCometDuration: time.Millisecond * 500,
DefaultHeaderCodec: "protobuf",
DefaultBodyCodec: "json",
DefaultBodyGzipLevel: 5,
PrintBody: false,
}
var peer = tp.NewPeer(cfg)
peer.PushRouter.Reg(new(Push))
{
var sess, err = peer.Dial("127.0.0.1:9090")
if err != nil {
tp.Panicf("%v", err)
}
var reply interface{}
var pullcmd = sess.Pull(
"/group/home/test?peer_id=client9090",
map[string]interface{}{"conn_port": 9090},
&reply,
)
if pullcmd.Xerror() != nil {
tp.Fatalf("pull error: %v", pullcmd.Xerror().Error())
}
tp.Infof("9090reply: %#v", reply)
}
{
var sess, err = peer.Dial("127.0.0.1:9091")
if err != nil {
tp.Panicf("%v", err)
}
var reply interface{}
var pullcmd = sess.Pull(
"/group/home/test_unknown?peer_id=client9091",
map[string]interface{}{"conn_port": 9091},
&reply,
)
if pullcmd.Xerror() != nil {
tp.Fatalf("pull error: %v", pullcmd.Xerror().Error())
}
tp.Infof("9091reply test_unknown: %#v", reply)
}
}
// Push controller
type Push struct {
tp.PushCtx
}
// Test handler
func (p *Push) Test(args *map[string]interface{}) {
tp.Infof("receive push(%s):\nargs: %#v\nquery: %#v\n", p.Ip(), args, p.Query())
}
Teleport 项目采用商业应用友好的 Apache2.0 协议发布
此处可能存在不合适展示的内容，页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容，可点击提交进行申诉，我们将尽快为您处理。