commons_go/server
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

ing

Browse Source
This commit is contained in:
crusader 2017-11-01 14:22:52 +09:00
parent 88d68b4ab3
commit 7e0de361a8
19 changed files with 168 additions and 644 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

127
client.go
View File

@ -1,127 +0,0 @@
package server
import (
"fmt"
"io"
"log"
"sync"
"sync/atomic"
"time"
"git.loafle.net/commons_go/logging"
)
type Client interface {
Start()
Stop()
}
func NewClient(ch ClientHandler) Client {
s := &client{
ch: ch,
}
return s
}
type client struct {
ch ClientHandler
Stats ConnStats
stopChan chan struct{}
stopWg sync.WaitGroup
}
func (c *client) Start() {
if nil == c.ch {
panic("Client: client handler must be specified.")
}
c.ch.Validate()
if c.stopChan != nil {
panic("Client: client is already running. Stop it before starting it again")
}
c.stopChan = make(chan struct{})
c.ch.OnStart()
c.stopWg.Add(1)
runClient(c)
}
func (c *client) Stop() {
if c.stopChan == nil {
panic("Client: client must be started before stopping it")
}
close(c.stopChan)
c.stopWg.Wait()
c.stopChan = nil
c.ch.OnStop()
}
func runClient(c *client) {
defer c.stopWg.Done()
var conn io.ReadWriteCloser
var err error
var stopping atomic.Value
dialChan := make(chan struct{})
go func() {
if conn, err = c.ch.Dial(); err != nil {
if stopping.Load() == nil {
logging.Logger.Error(fmt.Sprintf("Client: [%s].Cannot establish rpc connection: [%s]", c.ch.GetAddr(), err))
}
}
close(dialChan)
}()
select {
case <-c.stopChan:
stopping.Store(true)
<-dialChan
return
case <-dialChan:
c.Stats.incDialCalls()
}
if err != nil {
c.Stats.incDialErrors()
select {
case <-c.stopChan:
return
case <-time.After(time.Second):
}
return
}
go handleClientConnection(c, conn)
select {
case <-c.stopChan:
return
default:
}
}
func handleClientConnection(c *client, conn io.ReadWriteCloser) {
if err := c.ch.OnHandshake(c.ch.GetAddr(), conn); nil != err {
logging.Logger.Error(fmt.Sprintf("Client: [%s]. handshake error: [%s]", c.ch.GetAddr(), err))
conn.Close()
return
}
log.Printf("handleClientConnection")
clientStopChan := make(chan struct{})
go c.ch.Handle(conn, clientStopChan)
select {
case <-c.stopChan:
close(clientStopChan)
conn.Close()
return
case <-clientStopChan:
conn.Close()
return
}
}

24
client_handler.go
View File

@ -1,24 +0,0 @@
package server
import (
"io"
"time"
)
type ClientHandler interface {
OnStart()
OnStop()
Dial() (conn io.ReadWriteCloser, err error)
OnHandshake(remoteAddr string, rwc io.ReadWriteCloser) error
Handle(rwc io.ReadWriteCloser, stopChan chan struct{})
GetAddr() string
GetPendingRequests() int
GetRequestTimeout() time.Duration
GetSendBufferSize() int
GetRecvBufferSize() int
GetKeepAlivePeriod() time.Duration
Validate()
}

108
client_handlers.go
View File

@ -1,108 +0,0 @@
package server
import (
"errors"
"io"
"log"
"time"
)
type ClientHandlers struct {
// Server address to connect to.
//
// The address format depends on the underlying transport provided
// by Client.Dial. The following transports are provided out of the box:
// * TCP - see NewTCPClient() and NewTCPServer().
// * TLS - see NewTLSClient() and NewTLSServer().
// * Unix sockets - see NewUnixClient() and NewUnixServer().
//
// By default TCP transport is used.
Addr string
// The maximum number of pending requests in the queue.
//
// The number of pending requsts should exceed the expected number
// of concurrent goroutines calling client's methods.
// Otherwise a lot of ClientError.Overflow errors may appear.
//
// Default is DefaultPendingMessages.
PendingRequests int
// Maximum request time.
// Default value is DefaultRequestTimeout.
RequestTimeout time.Duration
// Size of send buffer per each underlying connection in bytes.
// Default value is DefaultBufferSize.
SendBufferSize int
// Size of recv buffer per each underlying connection in bytes.
// Default value is DefaultBufferSize.
RecvBufferSize int
KeepAlivePeriod time.Duration
}
func (ch *ClientHandlers) OnStart() {
// no op
}
func (ch *ClientHandlers) OnStop() {
// no op
}
func (ch *ClientHandlers) Dial() (conn io.ReadWriteCloser, err error) {
return nil, errors.New("Client: Handler method[Dial] of Client is not implement")
}
func (ch *ClientHandlers) OnHandshake(remoteAddr string, rwc io.ReadWriteCloser) error {
log.Printf("OnHandshake")
return nil
}
func (ch *ClientHandlers) Handle(rwc io.ReadWriteCloser, stopChan chan struct{}) {
// no op
log.Printf("Handle")
}
func (ch *ClientHandlers) GetAddr() string {
return ch.Addr
}
func (ch *ClientHandlers) GetPendingRequests() int {
return ch.PendingRequests
}
func (ch *ClientHandlers) GetRequestTimeout() time.Duration {
return ch.RequestTimeout
}
func (ch *ClientHandlers) GetSendBufferSize() int {
return ch.SendBufferSize
}
func (ch *ClientHandlers) GetRecvBufferSize() int {
return ch.RecvBufferSize
}
func (ch *ClientHandlers) GetKeepAlivePeriod() time.Duration {
return ch.KeepAlivePeriod
}
func (ch *ClientHandlers) Validate() {
if ch.PendingRequests <= 0 {
ch.PendingRequests = DefaultPendingMessages
}
if ch.RequestTimeout <= 0 {
ch.RequestTimeout = DefaultRequestTimeout
}
if ch.SendBufferSize <= 0 {
ch.SendBufferSize = DefaultBufferSize
}
if ch.RecvBufferSize <= 0 {
ch.RecvBufferSize = DefaultBufferSize
}
if ch.KeepAlivePeriod <= 0 {
ch.KeepAlivePeriod = DefaultKeepAlivePeriod
}
}

11
ipc/client_handlers.go
View File

@ -1,11 +0,0 @@
package ipc
import (
"git.loafle.net/commons_go/server"
)
type ClientHandlers struct {
server.ClientHandlers
path string
}

10
ipc/client_handlers_unix.go
View File

@ -1,10 +0,0 @@
package ipc
import (
"io"
"net"
)
func (ch *ClientHandlers) Dial() (conn io.ReadWriteCloser, err error) {
return net.Dial("unix", ch.Addr)
}

11
ipc/client_handlers_windows.go
View File

@ -1,11 +0,0 @@
package ipc
import (
"io"
"gopkg.in/natefinch/npipe.v2"
)
func (ch *ClientHandlers) Dial() (conn io.ReadWriteCloser, err error) {
return npipe.Dial(ch.Addr)
}

7
ipc/server_handlers.go
View File

@ -1,7 +0,0 @@
package ipc
import "git.loafle.net/commons_go/server"
type ServerHandlers struct {
server.ServerHandlers
}

20
ipc/server_handlers_unix.go
View File

@ -1,20 +0,0 @@
package ipc
import (
"net"
"os"
)
// Addr is ex:/tmp/server.sock
func (sh *ServerHandlers) Listen() (l net.Listener, err error) {
os.Remove(sh.Addr)
l, err = net.ListenUnix("unix", &net.UnixAddr{Name: sh.Addr, Net: "unix"})
os.Chmod(sh.Addr, 0777)
return
}
func (sh *ServerHandlers) OnStop() {
os.Remove(sh.Addr)
}

16
ipc/server_handlers_windows.go
View File

@ -1,16 +0,0 @@
package ipc
import (
"net"
"gopkg.in/natefinch/npipe.v2"
)
// Addr is ex:\\.\pipe\server
func (sh *ServerHandlers) Listen() (net.Listener, error) {
return npipe.Listen(s.path)
}
func (sh *ServerHandlers) OnStop() {
}

42
server.go
View File

@ -2,7 +2,6 @@ package server
import ( import (
"fmt" "fmt"
"io"
"net" "net"
"sync" "sync"
"sync/atomic" "sync/atomic"
@ -29,7 +28,7 @@ type server struct {
listener net.Listener listener net.Listener
Stats ConnStats Stats ServerStats
stopChan chan struct{} stopChan chan struct{}
stopWg sync.WaitGroup stopWg sync.WaitGroup
@ -44,16 +43,17 @@ func (s *server) Start() error {
if s.stopChan != nil { if s.stopChan != nil {
panic("Server: server is already running. Stop it before starting it again") panic("Server: server is already running. Stop it before starting it again")
} }
s.stopChan = make(chan struct{})
var err error var err error
if s.listener, err = s.sh.Listen(); nil != err { if s.listener, err = s.sh.Listen(); nil != err {
return err return err
} }
s.stopChan = make(chan struct{})
s.sh.OnStart() s.sh.OnStart()
s.stopWg.Add(1) s.stopWg.Add(1)
go runServer(s) go handleServer(s)
return nil return nil
} }
@ -76,21 +76,22 @@ func (s *server) Serve() error {
return nil return nil
} }
func runServer(s *server) { func handleServer(s *server) {
defer s.stopWg.Done() defer s.stopWg.Done()
var conn io.ReadWriteCloser var conn net.Conn
var clientAddr string
var err error var err error
var stopping atomic.Value var stopping atomic.Value
for { for {
acceptChan := make(chan struct{}) acceptChan := make(chan struct{})
go func() { go func() {
if conn, clientAddr, err = s.sh.accept(s.listener); err != nil { if conn, err = s.listener.Accept(); err != nil {
if stopping.Load() == nil { if stopping.Load() == nil {
logging.Logger.Error(fmt.Sprintf("Server: [%s]. Cannot accept new connection: [%s]", s.sh.GetAddr(), err)) logging.Logger.Error(fmt.Sprintf("Server: Cannot accept new connection: [%s]", err))
} }
} else {
conn, err = s.sh.OnAccept(conn)
} }
close(acceptChan) close(acceptChan)
}() }()
@ -116,31 +117,26 @@ func runServer(s *server) {
} }
s.stopWg.Add(1) s.stopWg.Add(1)
go handleServerConnection(s, conn, clientAddr) go handleConnection(s, conn)
} }
} }
func handleServerConnection(s *server, conn io.ReadWriteCloser, clientAddr string) { func handleConnection(s *server, conn net.Conn) {
defer s.stopWg.Done() defer s.stopWg.Done()
if err := s.sh.OnHandshake(clientAddr, conn); nil != err { logging.Logger.Debug(fmt.Sprintf("Server: Client[%s] is connected.", conn.RemoteAddr()))
logging.Logger.Error(fmt.Sprintf("Server: [%s]. handshake error: [%s]", clientAddr, err))
conn.Close()
return
}
logging.Logger.Debug(fmt.Sprintf("Server: Client[%s] is connected.", clientAddr))
clientStopChan := make(chan struct{}) clientStopChan := make(chan struct{})
go s.sh.Handle(clientAddr, conn, clientStopChan) handleDoneCnah := make(chan struct{})
go s.sh.Handle(conn, clientStopChan, handleDoneCnah)
select { select {
case <-s.stopChan: case <-s.stopChan:
close(clientStopChan) close(clientStopChan)
conn.Close() conn.Close()
return <-handleDoneCnah
case <-clientStopChan: case <-handleDoneCnah:
close(clientStopChan)
conn.Close() conn.Close()
logging.Logger.Debug(fmt.Sprintf("Server: Client[%s] is disconnected.", clientAddr)) logging.Logger.Debug(fmt.Sprintf("Server: Client[%s] is disconnected.", conn.RemoteAddr()))
return
} }
} }

12
server_handler.go
View File

@ -1,23 +1,19 @@
package server package server
import ( import (
"io"
"net" "net"
) )
type ServerHandler interface { type ServerHandler interface {
Listen() (net.Listener, error)
OnAccept(conn net.Conn) (net.Conn, error)
OnStart() OnStart()
OnStop() OnStop()
Listen() (net.Listener, error) Handle(conn net.Conn, stopChan <-chan struct{}, doneChan chan<- struct{})
OnHandshake(remoteAddr string, rwc io.ReadWriteCloser) error
Handle(remoteAddr string, rwc io.ReadWriteCloser, stopChan chan struct{})
IsClientDisconnect(err error) bool IsClientDisconnect(err error) bool
GetAddr() string
Validate() Validate()
accept(l net.Listener) (io.ReadWriteCloser, string, error)
} }

58
server_handlers.go
View File

@ -4,23 +4,9 @@ import (
"errors" "errors"
"io" "io"
"net" "net"
"time"
) )
type ServerHandlers struct { type ServerHandlers struct {
// Address to listen to for incoming connections.
//
// The address format depends on the underlying transport provided
// by Server.Listener. The following transports are provided
// out of the box:
// * TCP - see NewTCPServer() and NewTCPClient().
// * TLS (aka SSL) - see NewTLSServer() and NewTLSClient().
// * Unix sockets - see NewUnixServer() and NewUnixClient().
//
// By default TCP transport is used.
Addr string
KeepAlivePeriod time.Duration
} }
func (sh *ServerHandlers) OnStart() { func (sh *ServerHandlers) OnStart() {
@ -35,11 +21,11 @@ func (sh *ServerHandlers) Listen() (net.Listener, error) {
return nil, errors.New("Server: Handler method[Listen] of Server is not implement") return nil, errors.New("Server: Handler method[Listen] of Server is not implement")
} }
func (sh *ServerHandlers) OnHandshake(remoteAddr string, rwc io.ReadWriteCloser) error { func (sh *ServerHandlers) OnAccept(conn net.Conn) (net.Conn, error) {
return nil return conn, nil
} }
func (sh *ServerHandlers) Handle(remoteAddr string, rwc io.ReadWriteCloser, stopChan chan struct{}) { func (sh *ServerHandlers) Handle(conn net.Conn, stopChan <-chan struct{}, doneChan chan<- struct{}) {
} }
@ -47,44 +33,6 @@ func (sh *ServerHandlers) IsClientDisconnect(err error) bool {
return err == io.ErrUnexpectedEOF || err == io.EOF return err == io.ErrUnexpectedEOF || err == io.EOF
} }
func (sh *ServerHandlers) GetAddr() string {
return sh.Addr
}
func (sh *ServerHandlers) accept(l net.Listener) (io.ReadWriteCloser, string, error) {
conn, err := l.Accept()
if nil != err {
return nil, "", err
}
if 0 < sh.KeepAlivePeriod {
if err = setupKeepalive(conn, sh.KeepAlivePeriod); nil != err {
conn.Close()
return nil, "", err
}
}
return conn, conn.RemoteAddr().String(), nil
}
func (sh *ServerHandlers) Validate() { func (sh *ServerHandlers) Validate() {
if sh.KeepAlivePeriod <= 0 {
sh.KeepAlivePeriod = DefaultKeepAlivePeriod
}
} }
func setupKeepalive(conn net.Conn, keepAlivePeriod time.Duration) error {
tcpConn, ok := conn.(*net.TCPConn)
if !ok {
return errors.New("Server: Keepalive is valid when connection is net.TCPConn")
}
if err := tcpConn.SetKeepAlive(true); err != nil {
return err
}
if err := tcpConn.SetKeepAlivePeriod(keepAlivePeriod * time.Second); err != nil {
return err
}
return nil
}

44
stats.go
View File

@ -6,7 +6,7 @@ import (
"time" "time"
) )
type ConnStats struct { type ServerStats struct {
// The number of request performed. // The number of request performed.
RequestCount uint64 RequestCount uint64
@ -53,68 +53,68 @@ type ConnStats struct {
// AvgRequestTime returns the average Request execution time. // AvgRequestTime returns the average Request execution time.
// //
// Use stats returned from ConnStats.Snapshot() on live Client and / or Server, // Use stats returned from ServerStats.Snapshot() on live Client and / or Server,
// since the original stats can be updated by concurrently running goroutines. // since the original stats can be updated by concurrently running goroutines.
func (cs *ConnStats) AvgRequestTime() time.Duration { func (ss *ServerStats) AvgRequestTime() time.Duration {
return time.Duration(float64(cs.RequestTime)/float64(cs.RequestCount)) * time.Millisecond return time.Duration(float64(ss.RequestTime)/float64(ss.RequestCount)) * time.Millisecond
} }
// AvgRequestBytes returns the average bytes sent / received per Request. // AvgRequestBytes returns the average bytes sent / received per Request.
// //
// Use stats returned from ConnStats.Snapshot() on live Client and / or Server, // Use stats returned from ServerStats.Snapshot() on live Client and / or Server,
// since the original stats can be updated by concurrently running goroutines. // since the original stats can be updated by concurrently running goroutines.
func (cs *ConnStats) AvgRequestBytes() (send float64, recv float64) { func (ss *ServerStats) AvgRequestBytes() (send float64, recv float64) {
return float64(cs.BytesWritten) / float64(cs.RequestCount), float64(cs.BytesRead) / float64(cs.RequestCount) return float64(ss.BytesWritten) / float64(ss.RequestCount), float64(ss.BytesRead) / float64(ss.RequestCount)
} }
// AvgRequestCalls returns the average number of write() / read() syscalls per Request. // AvgRequestCalls returns the average number of write() / read() syscalls per Request.
// //
// Use stats returned from ConnStats.Snapshot() on live Client and / or Server, // Use stats returned from ServerStats.Snapshot() on live Client and / or Server,
// since the original stats can be updated by concurrently running goroutines. // since the original stats can be updated by concurrently running goroutines.
func (cs *ConnStats) AvgRequestCalls() (write float64, read float64) { func (ss *ServerStats) AvgRequestCalls() (write float64, read float64) {
return float64(cs.WriteCalls) / float64(cs.RequestCount), float64(cs.ReadCalls) / float64(cs.RequestCount) return float64(ss.WriteCalls) / float64(ss.RequestCount), float64(ss.ReadCalls) / float64(ss.RequestCount)
} }
type writerCounter struct { type writerCounter struct {
w io.Writer w io.Writer
cs *ConnStats ss *ServerStats
} }
type readerCounter struct { type readerCounter struct {
r io.Reader r io.Reader
cs *ConnStats ss *ServerStats
} }
func newWriterCounter(w io.Writer, cs *ConnStats) io.Writer { func newWriterCounter(w io.Writer, ss *ServerStats) io.Writer {
return &writerCounter{ return &writerCounter{
w: w, w: w,
cs: cs, ss: ss,
} }
} }
func newReaderCounter(r io.Reader, cs *ConnStats) io.Reader { func newReaderCounter(r io.Reader, ss *ServerStats) io.Reader {
return &readerCounter{ return &readerCounter{
r: r, r: r,
cs: cs, ss: ss,
} }
} }
func (w *writerCounter) Write(p []byte) (int, error) { func (w *writerCounter) Write(p []byte) (int, error) {
n, err := w.w.Write(p) n, err := w.w.Write(p)
w.cs.incWriteCalls() w.ss.incWriteCalls()
if err != nil { if err != nil {
w.cs.incWriteErrors() w.ss.incWriteErrors()
} }
w.cs.addBytesWritten(uint64(n)) w.ss.addBytesWritten(uint64(n))
return n, err return n, err
} }
func (r *readerCounter) Read(p []byte) (int, error) { func (r *readerCounter) Read(p []byte) (int, error) {
n, err := r.r.Read(p) n, err := r.r.Read(p)
r.cs.incReadCalls() r.ss.incReadCalls()
if err != nil { if err != nil {
r.cs.incReadErrors() r.ss.incReadErrors()
} }
r.cs.addBytesRead(uint64(n)) r.ss.addBytesRead(uint64(n))
return n, err return n, err
} }

136
stats_386.go
View File

@ -4,103 +4,103 @@ import "sync"
// Snapshot returns connection statistics' snapshot. // Snapshot returns connection statistics' snapshot.
// //
// Use stats returned from ConnStats.Snapshot() on live Client and / or Server, // Use stats returned from ServerStats.Snapshot() on live Client and / or Server,
// since the original stats can be updated by concurrently running goroutines. // since the original stats can be updated by concurrently running goroutines.
func (cs *ConnStats) Snapshot() *ConnStats { func (ss *ServerStats) Snapshot() *ServerStats {
cs.lock.Lock() ss.lock.Lock()
snapshot := *cs snapshot := *ss
cs.lock.Unlock() ss.lock.Unlock()
snapshot.lock = sync.Mutex{} snapshot.lock = sync.Mutex{}
return &snapshot return &snapshot
} }
// Reset resets all the stats counters. // Reset resets all the stats counters.
func (cs *ConnStats) Reset() { func (ss *ServerStats) Reset() {
cs.lock.Lock() ss.lock.Lock()
cs.RequestCount = 0 ss.RequestCount = 0
cs.RequestTime = 0 ss.RequestTime = 0
cs.BytesWritten = 0 ss.BytesWritten = 0
cs.BytesRead = 0 ss.BytesRead = 0
cs.WriteCalls = 0 ss.WriteCalls = 0
cs.WriteErrors = 0 ss.WriteErrors = 0
cs.ReadCalls = 0 ss.ReadCalls = 0
cs.ReadErrors = 0 ss.ReadErrors = 0
cs.DialCalls = 0 ss.DialCalls = 0
cs.DialErrors = 0 ss.DialErrors = 0
cs.AcceptCalls = 0 ss.AcceptCalls = 0
cs.AcceptErrors = 0 ss.AcceptErrors = 0
cs.lock.Unlock() ss.lock.Unlock()
} }
func (cs *ConnStats) incRPCCalls() { func (ss *ServerStats) incRPCCalls() {
cs.lock.Lock() ss.lock.Lock()
cs.RequestCount++ ss.RequestCount++
cs.lock.Unlock() ss.lock.Unlock()
} }
func (cs *ConnStats) incRPCTime(dt uint64) { func (ss *ServerStats) incRPCTime(dt uint64) {
cs.lock.Lock() ss.lock.Lock()
cs.RequestTime += dt ss.RequestTime += dt
cs.lock.Unlock() ss.lock.Unlock()
} }
func (cs *ConnStats) addBytesWritten(n uint64) { func (ss *ServerStats) addBytesWritten(n uint64) {
cs.lock.Lock() ss.lock.Lock()
cs.BytesWritten += n ss.BytesWritten += n
cs.lock.Unlock() ss.lock.Unlock()
} }
func (cs *ConnStats) addBytesRead(n uint64) { func (ss *ServerStats) addBytesRead(n uint64) {
cs.lock.Lock() ss.lock.Lock()
cs.BytesRead += n ss.BytesRead += n
cs.lock.Unlock() ss.lock.Unlock()
} }
func (cs *ConnStats) incReadCalls() { func (ss *ServerStats) incReadCalls() {
cs.lock.Lock() ss.lock.Lock()
cs.ReadCalls++ ss.ReadCalls++
cs.lock.Unlock() ss.lock.Unlock()
} }
func (cs *ConnStats) incReadErrors() { func (ss *ServerStats) incReadErrors() {
cs.lock.Lock() ss.lock.Lock()
cs.ReadErrors++ ss.ReadErrors++
cs.lock.Unlock() ss.lock.Unlock()
} }
func (cs *ConnStats) incWriteCalls() { func (ss *ServerStats) incWriteCalls() {
cs.lock.Lock() ss.lock.Lock()
cs.WriteCalls++ ss.WriteCalls++
cs.lock.Unlock() ss.lock.Unlock()
} }
func (cs *ConnStats) incWriteErrors() { func (ss *ServerStats) incWriteErrors() {
cs.lock.Lock() ss.lock.Lock()
cs.WriteErrors++ ss.WriteErrors++
cs.lock.Unlock() ss.lock.Unlock()
} }
func (cs *ConnStats) incDialCalls() { func (ss *ServerStats) incDialCalls() {
cs.lock.Lock() ss.lock.Lock()
cs.DialCalls++ ss.DialCalls++
cs.lock.Unlock() ss.lock.Unlock()
} }
func (cs *ConnStats) incDialErrors() { func (ss *ServerStats) incDialErrors() {
cs.lock.Lock() ss.lock.Lock()
cs.DialErrors++ ss.DialErrors++
cs.lock.Unlock() ss.lock.Unlock()
} }
func (cs *ConnStats) incAcceptCalls() { func (ss *ServerStats) incAcceptCalls() {
cs.lock.Lock() ss.lock.Lock()
cs.AcceptCalls++ ss.AcceptCalls++
cs.lock.Unlock() ss.lock.Unlock()
} }
func (cs *ConnStats) incAcceptErrors() { func (ss *ServerStats) incAcceptErrors() {
cs.lock.Lock() ss.lock.Lock()
cs.AcceptErrors++ ss.AcceptErrors++
cs.lock.Unlock() ss.lock.Unlock()
} }

104
stats_generic.go
View File

@ -4,85 +4,85 @@ import "sync/atomic"
// Snapshot returns connection statistics' snapshot. // Snapshot returns connection statistics' snapshot.
// //
// Use stats returned from ConnStats.Snapshot() on live Client and / or Server, // Use stats returned from ServerStats.Snapshot() on live Client and / or Server,
// since the original stats can be updated by concurrently running goroutines. // since the original stats can be updated by concurrently running goroutines.
func (cs *ConnStats) Snapshot() *ConnStats { func (ss *ServerStats) Snapshot() *ServerStats {
return &ConnStats{ return &ServerStats{
RequestCount: atomic.LoadUint64(&cs.RequestCount), RequestCount: atomic.LoadUint64(&ss.RequestCount),
RequestTime: atomic.LoadUint64(&cs.RequestTime), RequestTime: atomic.LoadUint64(&ss.RequestTime),
BytesWritten: atomic.LoadUint64(&cs.BytesWritten), BytesWritten: atomic.LoadUint64(&ss.BytesWritten),
BytesRead: atomic.LoadUint64(&cs.BytesRead), BytesRead: atomic.LoadUint64(&ss.BytesRead),
ReadCalls: atomic.LoadUint64(&cs.ReadCalls), ReadCalls: atomic.LoadUint64(&ss.ReadCalls),
ReadErrors: atomic.LoadUint64(&cs.ReadErrors), ReadErrors: atomic.LoadUint64(&ss.ReadErrors),
WriteCalls: atomic.LoadUint64(&cs.WriteCalls), WriteCalls: atomic.LoadUint64(&ss.WriteCalls),
WriteErrors: atomic.LoadUint64(&cs.WriteErrors), WriteErrors: atomic.LoadUint64(&ss.WriteErrors),
DialCalls: atomic.LoadUint64(&cs.DialCalls), DialCalls: atomic.LoadUint64(&ss.DialCalls),
DialErrors: atomic.LoadUint64(&cs.DialErrors), DialErrors: atomic.LoadUint64(&ss.DialErrors),
AcceptCalls: atomic.LoadUint64(&cs.AcceptCalls), AcceptCalls: atomic.LoadUint64(&ss.AcceptCalls),
AcceptErrors: atomic.LoadUint64(&cs.AcceptErrors), AcceptErrors: atomic.LoadUint64(&ss.AcceptErrors),
} }
} }
// Reset resets all the stats counters. // Reset resets all the stats counters.
func (cs *ConnStats) Reset() { func (ss *ServerStats) Reset() {
atomic.StoreUint64(&cs.RequestCount, 0) atomic.StoreUint64(&ss.RequestCount, 0)
atomic.StoreUint64(&cs.RequestTime, 0) atomic.StoreUint64(&ss.RequestTime, 0)
atomic.StoreUint64(&cs.BytesWritten, 0) atomic.StoreUint64(&ss.BytesWritten, 0)
atomic.StoreUint64(&cs.BytesRead, 0) atomic.StoreUint64(&ss.BytesRead, 0)
atomic.StoreUint64(&cs.WriteCalls, 0) atomic.StoreUint64(&ss.WriteCalls, 0)
atomic.StoreUint64(&cs.WriteErrors, 0) atomic.StoreUint64(&ss.WriteErrors, 0)
atomic.StoreUint64(&cs.ReadCalls, 0) atomic.StoreUint64(&ss.ReadCalls, 0)
atomic.StoreUint64(&cs.ReadErrors, 0) atomic.StoreUint64(&ss.ReadErrors, 0)
atomic.StoreUint64(&cs.DialCalls, 0) atomic.StoreUint64(&ss.DialCalls, 0)
atomic.StoreUint64(&cs.DialErrors, 0) atomic.StoreUint64(&ss.DialErrors, 0)
atomic.StoreUint64(&cs.AcceptCalls, 0) atomic.StoreUint64(&ss.AcceptCalls, 0)
atomic.StoreUint64(&cs.AcceptErrors, 0) atomic.StoreUint64(&ss.AcceptErrors, 0)
} }
func (cs *ConnStats) incRPCCalls() { func (ss *ServerStats) incRPCCalls() {
atomic.AddUint64(&cs.RequestCount, 1) atomic.AddUint64(&ss.RequestCount, 1)
} }
func (cs *ConnStats) incRPCTime(dt uint64) { func (ss *ServerStats) incRPCTime(dt uint64) {
atomic.AddUint64(&cs.RequestTime, dt) atomic.AddUint64(&ss.RequestTime, dt)
} }
func (cs *ConnStats) addBytesWritten(n uint64) { func (ss *ServerStats) addBytesWritten(n uint64) {
atomic.AddUint64(&cs.BytesWritten, n) atomic.AddUint64(&ss.BytesWritten, n)
} }
func (cs *ConnStats) addBytesRead(n uint64) { func (ss *ServerStats) addBytesRead(n uint64) {
atomic.AddUint64(&cs.BytesRead, n) atomic.AddUint64(&ss.BytesRead, n)
} }
func (cs *ConnStats) incReadCalls() { func (ss *ServerStats) incReadCalls() {
atomic.AddUint64(&cs.ReadCalls, 1) atomic.AddUint64(&ss.ReadCalls, 1)
} }
func (cs *ConnStats) incReadErrors() { func (ss *ServerStats) incReadErrors() {
atomic.AddUint64(&cs.ReadErrors, 1) atomic.AddUint64(&ss.ReadErrors, 1)
} }
func (cs *ConnStats) incWriteCalls() { func (ss *ServerStats) incWriteCalls() {
atomic.AddUint64(&cs.WriteCalls, 1) atomic.AddUint64(&ss.WriteCalls, 1)
} }
func (cs *ConnStats) incWriteErrors() { func (ss *ServerStats) incWriteErrors() {
atomic.AddUint64(&cs.WriteErrors, 1) atomic.AddUint64(&ss.WriteErrors, 1)
} }
func (cs *ConnStats) incDialCalls() { func (ss *ServerStats) incDialCalls() {
atomic.AddUint64(&cs.DialCalls, 1) atomic.AddUint64(&ss.DialCalls, 1)
} }
func (cs *ConnStats) incDialErrors() { func (ss *ServerStats) incDialErrors() {
atomic.AddUint64(&cs.DialErrors, 1) atomic.AddUint64(&ss.DialErrors, 1)
} }
func (cs *ConnStats) incAcceptCalls() { func (ss *ServerStats) incAcceptCalls() {
atomic.AddUint64(&cs.AcceptCalls, 1) atomic.AddUint64(&ss.AcceptCalls, 1)
} }
func (cs *ConnStats) incAcceptErrors() { func (ss *ServerStats) incAcceptErrors() {
atomic.AddUint64(&cs.AcceptErrors, 1) atomic.AddUint64(&ss.AcceptErrors, 1)
} }

22
tcp/client_handlers.go
View File

@ -1,22 +0,0 @@
package tcp
import (
"io"
"net"
"time"
"git.loafle.net/commons_go/server"
)
type ClientHandlers struct {
server.ClientHandlers
}
func (ch *ClientHandlers) Dial() (conn io.ReadWriteCloser, err error) {
dialer := &net.Dialer{
Timeout: ch.RequestTimeout * time.Second,
KeepAlive: ch.KeepAlivePeriod * time.Second,
}
return dialer.Dial("tcp", ch.Addr)
}

16
tcp/server_handlers.go
View File

@ -1,16 +0,0 @@
package tcp
import (
"net"
"git.loafle.net/commons_go/server"
)
type ServerHandlers struct {
server.ServerHandlers
}
func (sh *ServerHandlers) Listen() (net.Listener, error) {
return net.Listen("tcp", sh.Addr)
}

25
tcp_tls/client_handlers.go
View File

@ -1,25 +0,0 @@
package tcp_tls
import (
"crypto/tls"
"io"
"net"
"time"
"git.loafle.net/commons_go/server"
)
type ClientHandlers struct {
server.ClientHandlers
tlsConfig *tls.Config
}
func (ch *ClientHandlers) Dial() (conn io.ReadWriteCloser, err error) {
dialer := &net.Dialer{
Timeout: ch.RequestTimeout * time.Second,
KeepAlive: ch.KeepAlivePeriod * time.Second,
}
return tls.DialWithDialer(dialer, "tcp", ch.Addr, ch.tlsConfig)
}

19
tcp_tls/server_handlers.go
View File

@ -1,19 +0,0 @@
package tcp_tls
import (
"crypto/tls"
"net"
"git.loafle.net/commons_go/server"
)
type ServerHandlers struct {
server.ServerHandlers
tlsConfig *tls.Config
}
func (sh *ServerHandlers) Listen() (net.Listener, error) {
return tls.Listen("tcp", sh.Addr, sh.tlsConfig)
}