certusone
/
tendermint
mirror of https://github.com/certusone/tendermint.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

import go-common as cmn

This commit is contained in:
Anton Kaliaev 2017-04-04 19:28:28 +04:00
parent 97a5ed2d1a
commit 868017cf1a
No known key found for this signature in database
GPG Key ID: 7B6881D965918214
1 changed files with 19 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
connection.go
View File

@ -10,10 +10,10 @@ import (
"sync/atomic" "sync/atomic"
"time" "time"
. "github.com/tendermint/go-common" cmn "github.com/tendermint/go-common"
cfg "github.com/tendermint/go-config" cfg "github.com/tendermint/go-config"
flow "github.com/tendermint/go-flowrate/flowrate" flow "github.com/tendermint/go-flowrate/flowrate"
"github.com/tendermint/go-wire" //"github.com/tendermint/log15" wire "github.com/tendermint/go-wire"
) )
const ( const (
@ -60,7 +60,7 @@ queue is full.
Inbound message bytes are handled with an onReceive callback function. Inbound message bytes are handled with an onReceive callback function.
*/ */
type MConnection struct { type MConnection struct {
BaseService cmn.BaseService
conn net.Conn conn net.Conn
bufReader *bufio.Reader bufReader *bufio.Reader
@ -78,9 +78,9 @@ type MConnection struct {
errored uint32 errored uint32
quit chan struct{} quit chan struct{}
flushTimer *ThrottleTimer // flush writes as necessary but throttled. flushTimer *cmn.ThrottleTimer // flush writes as necessary but throttled.
pingTimer *RepeatTimer // send pings periodically pingTimer *cmn.RepeatTimer // send pings periodically
chStatsTimer *RepeatTimer // update channel stats periodically chStatsTimer *cmn.RepeatTimer // update channel stats periodically
LocalAddress *NetAddress LocalAddress *NetAddress
RemoteAddress *NetAddress RemoteAddress *NetAddress
@ -123,7 +123,7 @@ func NewMConnection(config cfg.Config, conn net.Conn, chDescs []*ChannelDescript
mconn.channels = channels mconn.channels = channels
mconn.channelsIdx = channelsIdx mconn.channelsIdx = channelsIdx
mconn.BaseService = *NewBaseService(log, "MConnection", mconn) mconn.BaseService = *cmn.NewBaseService(log, "MConnection", mconn)
return mconn return mconn
} }
@ -131,9 +131,9 @@ func NewMConnection(config cfg.Config, conn net.Conn, chDescs []*ChannelDescript
func (c *MConnection) OnStart() error { func (c *MConnection) OnStart() error {
c.BaseService.OnStart() c.BaseService.OnStart()
c.quit = make(chan struct{}) c.quit = make(chan struct{})
c.flushTimer = NewThrottleTimer("flush", flushThrottleMS*time.Millisecond) c.flushTimer = cmn.NewThrottleTimer("flush", flushThrottleMS*time.Millisecond)
c.pingTimer = NewRepeatTimer("ping", pingTimeoutSeconds*time.Second) c.pingTimer = cmn.NewRepeatTimer("ping", pingTimeoutSeconds*time.Second)
c.chStatsTimer = NewRepeatTimer("chStats", updateStatsSeconds*time.Second) c.chStatsTimer = cmn.NewRepeatTimer("chStats", updateStatsSeconds*time.Second)
go c.sendRoutine() go c.sendRoutine()
go c.recvRoutine() go c.recvRoutine()
return nil return nil
@ -171,7 +171,7 @@ func (c *MConnection) flush() {
func (c *MConnection) _recover() { func (c *MConnection) _recover() {
if r := recover(); r != nil { if r := recover(); r != nil {
stack := debug.Stack() stack := debug.Stack()
err := StackError{r, stack} err := cmn.StackError{r, stack}
c.stopForError(err) c.stopForError(err)
} }
} }
@ -196,7 +196,7 @@ func (c *MConnection) Send(chID byte, msg interface{}) bool {
// Send message to channel. // Send message to channel.
channel, ok := c.channelsIdx[chID] channel, ok := c.channelsIdx[chID]
if !ok { if !ok {
log.Error(Fmt("Cannot send bytes, unknown channel %X", chID)) log.Error(cmn.Fmt("Cannot send bytes, unknown channel %X", chID))
return false return false
} }
@ -225,7 +225,7 @@ func (c *MConnection) TrySend(chID byte, msg interface{}) bool {
// Send message to channel. // Send message to channel.
channel, ok := c.channelsIdx[chID] channel, ok := c.channelsIdx[chID]
if !ok { if !ok {
log.Error(Fmt("Cannot send bytes, unknown channel %X", chID)) log.Error(cmn.Fmt("Cannot send bytes, unknown channel %X", chID))
return false return false
} }
@ -248,7 +248,7 @@ func (c *MConnection) CanSend(chID byte) bool {
channel, ok := c.channelsIdx[chID] channel, ok := c.channelsIdx[chID]
if !ok { if !ok {
log.Error(Fmt("Unknown channel %X", chID)) log.Error(cmn.Fmt("Unknown channel %X", chID))
return false return false
} }
return channel.canSend() return channel.canSend()
@ -424,7 +424,7 @@ FOR_LOOP:
} }
channel, ok := c.channelsIdx[pkt.ChannelID] channel, ok := c.channelsIdx[pkt.ChannelID]
if !ok || channel == nil { if !ok || channel == nil {
PanicQ(Fmt("Unknown channel %X", pkt.ChannelID)) cmn.PanicQ(cmn.Fmt("Unknown channel %X", pkt.ChannelID))
} }
msgBytes, err := channel.recvMsgPacket(pkt) msgBytes, err := channel.recvMsgPacket(pkt)
if err != nil { if err != nil {
@ -439,7 +439,7 @@ FOR_LOOP:
c.onReceive(pkt.ChannelID, msgBytes) c.onReceive(pkt.ChannelID, msgBytes)
} }
default: default:
PanicSanity(Fmt("Unknown message type %X", pktType)) cmn.PanicSanity(cmn.Fmt("Unknown message type %X", pktType))
} }
// TODO: shouldn't this go in the sendRoutine? // TODO: shouldn't this go in the sendRoutine?
@ -524,7 +524,7 @@ type Channel struct {
func newChannel(conn *MConnection, desc *ChannelDescriptor) *Channel { func newChannel(conn *MConnection, desc *ChannelDescriptor) *Channel {
desc.FillDefaults() desc.FillDefaults()
if desc.Priority <= 0 { if desc.Priority <= 0 {
PanicSanity("Channel default priority must be a postive integer") cmn.PanicSanity("Channel default priority must be a postive integer")
} }
return &Channel{ return &Channel{
conn: conn, conn: conn,
@ -593,14 +593,14 @@ func (ch *Channel) isSendPending() bool {
func (ch *Channel) nextMsgPacket() msgPacket { func (ch *Channel) nextMsgPacket() msgPacket {
packet := msgPacket{} packet := msgPacket{}
packet.ChannelID = byte(ch.id) packet.ChannelID = byte(ch.id)
packet.Bytes = ch.sending[:MinInt(maxMsgPacketPayloadSize, len(ch.sending))] packet.Bytes = ch.sending[:cmn.MinInt(maxMsgPacketPayloadSize, len(ch.sending))]
if len(ch.sending) <= maxMsgPacketPayloadSize { if len(ch.sending) <= maxMsgPacketPayloadSize {
packet.EOF = byte(0x01) packet.EOF = byte(0x01)
ch.sending = nil ch.sending = nil
atomic.AddInt32(&ch.sendQueueSize, -1) // decrement sendQueueSize atomic.AddInt32(&ch.sendQueueSize, -1) // decrement sendQueueSize
} else { } else {
packet.EOF = byte(0x00) packet.EOF = byte(0x00)
ch.sending = ch.sending[MinInt(maxMsgPacketPayloadSize, len(ch.sending)):] ch.sending = ch.sending[cmn.MinInt(maxMsgPacketPayloadSize, len(ch.sending)):]
} }
return packet return packet
} }