p2pNet

Documentation

Index

• Constants
• func ReadData(online *OnlineNode, ctx context.Context)
• func StreamToRw(s network.Stream) (*bufio.Reader, *bufio.Writer)
• func WriteData(rw *bufio.ReadWriter, data P2pNetMessageReceiveInterface) error
• type HearBeat
  • func NewHearBeat() *HearBeat
  • func (h *HearBeat) ReceiveHandle(ctx context.Context, node *OnlineNode)
• type OnlineNode
  • func IdToOnline(usrNode *UserNode, id string, ctx context.Context) (*OnlineNode, error)
  • func NewOnlineNode(usrNode *UserNode, stream network.Stream, isInOnlineNodes bool) *OnlineNode
  • func (oln *OnlineNode) AddOnlineNodesTry()
  • func (oln *OnlineNode) Close()
  • func (oln *OnlineNode) GetIdPretty() string
  • func (oln *OnlineNode) ListenRead()
  • func (oln *OnlineNode) WriteData(data P2pNetMessageReceiveInterface) error
• type P2pNetMessage
  • func (p *P2pNetMessage) PushHandle(rw *bufio.ReadWriter) error
  • func (p *P2pNetMessage) ReceiveHandle(ctx context.Context, node *OnlineNode)
• type P2pNetMessagePushInterface
  • func NewP2pNetMessageEncode(data P2pNetMessageReceiveInterface) (P2pNetMessagePushInterface, error)
• type P2pNetMessageReceiveInterface
  • func NewP2pNetMessageDecode(rw *bufio.ReadWriter) (P2pNetMessageReceiveInterface, error)
• type UserNode
  • func NewUserNode(port int, userKey, userData string) *UserNode
  • func (usr *UserNode) CheckOnlineNodes()
  • func (usr *UserNode) Connect(pi peer.AddrInfo) error
  • func (usr *UserNode) ConnectP2p() error
  • func (usr *UserNode) EachOnlineNodes(fn func(node *OnlineNode) bool)
  • func (usr *UserNode) Exit()
  • func (usr *UserNode) FindPeers() (<-chan peer.AddrInfo, error)
  • func (usr *UserNode) GetOnlineNode(id string) *OnlineNode
  • func (usr *UserNode) GetOnlineNodesCount() int
  • func (usr *UserNode) GetPriKey(key string) (*keys.PrivateKey, error)
  • func (usr *UserNode) IDPretty() string
  • func (usr *UserNode) IsLockedOnlineNodes() bool
  • func (usr *UserNode) LockedOnlineNode()
  • func (usr *UserNode) NewStream(pi peer.AddrInfo) (network.Stream, error)
  • func (usr *UserNode) NewStreamCtx(ctx context.Context, pi peer.AddrInfo) (network.Stream, error)
  • func (usr *UserNode) Ping(pi peer.ID, ctx context.Context) <-chan ping.Result
  • func (usr *UserNode) RemoveOnlineNode(id string)
  • func (usr *UserNode) RemoveOnlineNodeLocked(id string)
  • func (usr *UserNode) SignMsg(key string, msg string) (string, error)
  • func (usr *UserNode) SyncOnlineNodes()
  • func (usr *UserNode) UnLockedOnlineNode()
  • func (usr *UserNode) WaitOnlineNode() <-chan struct{}
  • func (usr *UserNode) WaitOnlineNodeLimitNum(num int) <-chan struct{}

Constants

const MaxOnlineNodesNum = 128

const NodePingTimeoutDuration = time.Second * 15

const OnlineNodesChkDuration = time.Minute * 5

const OnlineNodesSyncDuration = time.Second * 30

Functions

func ReadData

func ReadData(online *OnlineNode, ctx context.Context)

func StreamToRw

func StreamToRw(s network.Stream) (*bufio.Reader, *bufio.Writer)

func WriteData

func WriteData(rw *bufio.ReadWriter, data P2pNetMessageReceiveInterface) error

Types

type HearBeat

type HearBeat struct {
	Time time.Time
}

func NewHearBeat

func NewHearBeat() *HearBeat

func (*HearBeat) ReceiveHandle

func (h *HearBeat) ReceiveHandle(ctx context.Context, node *OnlineNode)

type OnlineNode

type OnlineNode struct {
	Pi              peer.AddrInfo
	Stream          network.Stream
	Rw              *bufio.ReadWriter
	LastChkTime     time.Time
	IsInOnlineNodes bool

	IsLister bool
	// contains filtered or unexported fields
}

func IdToOnline

func IdToOnline(usrNode *UserNode, id string, ctx context.Context) (*OnlineNode, error)

func NewOnlineNode

func NewOnlineNode(usrNode *UserNode, stream network.Stream, isInOnlineNodes bool) *OnlineNode

func (*OnlineNode) AddOnlineNodesTry

func (oln *OnlineNode) AddOnlineNodesTry()

func (*OnlineNode) Close

func (oln *OnlineNode) Close()

func (*OnlineNode) GetIdPretty

func (oln *OnlineNode) GetIdPretty() string

func (*OnlineNode) ListenRead

func (oln *OnlineNode) ListenRead()

func (*OnlineNode) WriteData

func (oln *OnlineNode) WriteData(data P2pNetMessageReceiveInterface) error

type P2pNetMessage

type P2pNetMessage struct {
	Msg       P2pNetMessageReceiveInterface
	CreatedAt time.Time
}

func (*P2pNetMessage) PushHandle

func (p *P2pNetMessage) PushHandle(rw *bufio.ReadWriter) error

func (*P2pNetMessage) ReceiveHandle

func (p *P2pNetMessage) ReceiveHandle(ctx context.Context, node *OnlineNode)

type P2pNetMessagePushInterface

type P2pNetMessagePushInterface interface {
	PushHandle(writer *bufio.ReadWriter) error
}

func NewP2pNetMessageEncode

func NewP2pNetMessageEncode(data P2pNetMessageReceiveInterface) (P2pNetMessagePushInterface, error)

type P2pNetMessageReceiveInterface

type P2pNetMessageReceiveInterface interface {
	ReceiveHandle(ctx context.Context, node *OnlineNode)
}

func NewP2pNetMessageDecode

func NewP2pNetMessageDecode(rw *bufio.ReadWriter) (P2pNetMessageReceiveInterface, error)

type UserNode

type UserNode struct {
	Ctx    context.Context    //p2p上下文
	Cancel context.CancelFunc //用户取消当前上下文 平滑退出

	Protocol string //当前节点协议
	Host     host.Host
	UserKey  string //存储私钥的key
	UserData string //存储私钥的目录

	Port int
	// contains filtered or unexported fields
}

func NewUserNode

func NewUserNode(port int, userKey, userData string) *UserNode

新建一个用户节点

func (*UserNode) CheckOnlineNodes

func (usr *UserNode) CheckOnlineNodes()

func (*UserNode) Connect

func (usr *UserNode) Connect(pi peer.AddrInfo) error

func (*UserNode) ConnectP2p

func (usr *UserNode) ConnectP2p() error

func (*UserNode) EachOnlineNodes

func (usr *UserNode) EachOnlineNodes(fn func(node *OnlineNode) bool)

func (*UserNode) Exit

func (usr *UserNode) Exit()

在这里处理退出流程

func (*UserNode) FindPeers

func (usr *UserNode) FindPeers() (<-chan peer.AddrInfo, error)

func (*UserNode) GetOnlineNode

func (usr *UserNode) GetOnlineNode(id string) *OnlineNode

func (*UserNode) GetOnlineNodesCount

func (usr *UserNode) GetOnlineNodesCount() int

func (*UserNode) GetPriKey

func (usr *UserNode) GetPriKey(key string) (*keys.PrivateKey, error)

获取用户私钥 没有则创建

func (*UserNode) IDPretty

func (usr *UserNode) IDPretty() string

func (*UserNode) IsLockedOnlineNodes

func (usr *UserNode) IsLockedOnlineNodes() bool

func (*UserNode) LockedOnlineNode

func (usr *UserNode) LockedOnlineNode()

func (*UserNode) NewStream

func (usr *UserNode) NewStream(pi peer.AddrInfo) (network.Stream, error)

func (*UserNode) NewStreamCtx

func (usr *UserNode) NewStreamCtx(ctx context.Context, pi peer.AddrInfo) (network.Stream, error)

func (*UserNode) Ping

func (usr *UserNode) Ping(pi peer.ID, ctx context.Context) <-chan ping.Result

func (*UserNode) RemoveOnlineNode

func (usr *UserNode) RemoveOnlineNode(id string)

func (*UserNode) RemoveOnlineNodeLocked

func (usr *UserNode) RemoveOnlineNodeLocked(id string)

func (*UserNode) SignMsg

func (usr *UserNode) SignMsg(key string, msg string) (string, error)

func (*UserNode) SyncOnlineNodes

func (usr *UserNode) SyncOnlineNodes()

func (*UserNode) UnLockedOnlineNode

func (usr *UserNode) UnLockedOnlineNode()

func (*UserNode) WaitOnlineNode

func (usr *UserNode) WaitOnlineNode() <-chan struct{}

检查并等待在线列表

func (*UserNode) WaitOnlineNodeLimitNum

func (usr *UserNode) WaitOnlineNodeLimitNum(num int) <-chan struct{}