ing

server.go

@@ -22,7 +22,7 @@ type server struct {
 
 	listener net.Listener
 
-	Stats ServerStats
+	Stats ConnStats
 
 	stopChan chan struct{}
 	stopWg   sync.WaitGroup
@@ -55,6 +55,7 @@ func (s *server) Stop() {
 	close(s.stopChan)
 	s.stopWg.Wait()
 	s.stopChan = nil
+	s.serverHandler.Stopped()
 }
 
 func (s *server) Serve() error {

server_handler.go

@@ -8,9 +8,9 @@ import (
 type ServerHandler interface {
 	Listen() (net.Listener, error)
 	Handle(remoteAddr string, rwc io.ReadWriteCloser, stopChan chan struct{})
+	Stopped()
 
 	GetAddr() string
-	GetPendingResponses() int
 
 	Validate()
 

server_handlers.go

@@ -43,27 +43,6 @@ type ServerHandlers struct {
 	// By default TCP transport is used.
 	Addr string
 
-	// The maximum delay between response flushes to clients.
-	//
-	// Negative values lead to immediate requests' sending to the client
-	// without their buffering. This minimizes rpc latency at the cost
-	// of higher CPU and network usage.
-	//
-	// Default is DefaultFlushDelay.
-	FlushDelay time.Duration
-
-	// The maximum number of pending responses in the queue.
-	// Default is DefaultPendingMessages.
-	PendingResponses int
-
-	// Size of send buffer per each underlying connection in bytes.
-	// Default is DefaultBufferSize.
-	SendBufferSize int
-
-	// Size of recv buffer per each underlying connection in bytes.
-	// Default is DefaultBufferSize.
-	RecvBufferSize int
-
 	KeepAlivePeriod time.Duration
 }
 
@@ -75,12 +54,12 @@ func (sh *ServerHandlers) Handle(remoteAddr string, rwc io.ReadWriteCloser, stop
 
 }
 
-func (sh *ServerHandlers) GetAddr() string {
+func (sh *ServerHandlers) Stopped() {
-	return sh.Addr
+
 }
 
-func (sh *ServerHandlers) GetPendingResponses() int {
+func (sh *ServerHandlers) GetAddr() string {
-	return sh.PendingResponses
+	return sh.Addr
 }
 
 // func (sh *ServerHandlers)  {
@@ -125,21 +104,6 @@ func (sh *ServerHandlers) accept(l net.Listener) (io.ReadWriteCloser, string, er
 }
 
 func (sh *ServerHandlers) Validate() {
-	if sh.Concurrency <= 0 {
-		sh.Concurrency = DefaultConcurrency
-	}
-	if sh.FlushDelay == 0 {
-		sh.FlushDelay = DefaultFlushDelay
-	}
-	if sh.PendingResponses <= 0 {
-		sh.PendingResponses = DefaultPendingMessages
-	}
-	if sh.SendBufferSize <= 0 {
-		sh.SendBufferSize = DefaultBufferSize
-	}
-	if sh.RecvBufferSize <= 0 {
-		sh.RecvBufferSize = DefaultBufferSize
-	}
 	if sh.KeepAlivePeriod <= 0 {
 		sh.KeepAlivePeriod = DefaultKeepAlivePeriod
 	}

stats.go

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

stats_386.go

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

stats_generic.go

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