integrate packageManager into servers
diff --git a/request-server.go b/request-server.go
index 95fdf3f..4d70373 100644
--- a/request-server.go
+++ b/request-server.go
@@ -1,6 +1,7 @@
package sftp
import (
+ "encoding"
"io"
"os"
"path/filepath"
@@ -28,6 +29,7 @@
serverConn
Handlers Handlers
pktChan chan requestPacket
+ pktMgr packetManager
openRequests map[string]Request
openRequestLock sync.RWMutex
handleCount int
@@ -36,15 +38,17 @@
// NewRequestServer creates/allocates/returns new RequestServer.
// Normally there there will be one server per user-session.
func NewRequestServer(rwc io.ReadWriteCloser, h Handlers) *RequestServer {
- return &RequestServer{
- serverConn: serverConn{
- conn: conn{
- Reader: rwc,
- WriteCloser: rwc,
- },
+ svrConn := serverConn{
+ conn: conn{
+ Reader: rwc,
+ WriteCloser: rwc,
},
+ }
+ return &RequestServer{
+ serverConn: svrConn,
Handlers: h,
pktChan: make(chan requestPacket, sftpServerWorkerCount),
+ pktMgr: newPktMgr(&svrConn),
openRequests: make(map[string]Request),
}
}
@@ -91,6 +95,7 @@
}
var err error
+ var pkt requestPacket
var pktType uint8
var pktBytes []byte
for {
@@ -98,15 +103,19 @@
if err != nil {
break
}
- pkt, err := makePacket(rxPacket{fxp(pktType), pktBytes})
+ pkt, err = makePacket(rxPacket{fxp(pktType), pktBytes})
if err != nil {
+ debug("makePacket err: %v", err)
+ rs.conn.Close() // shuts down recvPacket
break
}
+ rs.pktMgr.incomingPacket(pkt)
rs.pktChan <- pkt
}
close(rs.pktChan) // shuts down sftpServerWorkers
wg.Wait() // wait for all workers to exit
+ rs.pktMgr.close() // shuts down packetManager
return err
}
@@ -176,6 +185,20 @@
return rpkt
}
+// Wrap underlying connection methods to use packetManager
+func (rs RequestServer) sendPacket(m encoding.BinaryMarshaler) error {
+ if pkt, ok := m.(responsePacket); ok {
+ rs.pktMgr.readyPacket(pkt)
+ } else {
+ return errors.Errorf("unexpected packet type %T", m)
+ }
+ return nil
+}
+
+func (rs RequestServer) sendError(p ider, err error) error {
+ return rs.sendPacket(statusFromError(p, err))
+}
+
// os.ErrNotExist should convert to ssh_FX_NO_SUCH_FILE, but is not recognized
// by statusFromError. So we convert to syscall.ENOENT which it does.
func errorAdapter(err error) error {
diff --git a/server.go b/server.go
index f0bbb8b..cf200b1 100644
--- a/server.go
+++ b/server.go
@@ -30,6 +30,7 @@
debugStream io.Writer
readOnly bool
pktChan chan requestPacket
+ pktMgr packetManager
openFiles map[string]*os.File
openFilesLock sync.RWMutex
handleCount int
@@ -75,15 +76,17 @@
//
// A subsequent call to Serve() is required to begin serving files over SFTP.
func NewServer(rwc io.ReadWriteCloser, options ...ServerOption) (*Server, error) {
- s := &Server{
- serverConn: serverConn{
- conn: conn{
- Reader: rwc,
- WriteCloser: rwc,
- },
+ svrConn := serverConn{
+ conn: conn{
+ Reader: rwc,
+ WriteCloser: rwc,
},
+ }
+ s := &Server{
+ serverConn: svrConn,
debugStream: ioutil.Discard,
pktChan: make(chan requestPacket, sftpServerWorkerCount),
+ pktMgr: newPktMgr(&svrConn),
openFiles: make(map[string]*os.File),
maxTxPacket: 1 << 15,
}
@@ -303,15 +306,18 @@
pkt, err = makePacket(rxPacket{fxp(pktType), pktBytes})
if err != nil {
+ debug("makePacket err: %v", err)
svr.conn.Close() // shuts down recvPacket
break
}
+ svr.pktMgr.incomingPacket(pkt)
svr.pktChan <- pkt
}
close(svr.pktChan) // shuts down sftpServerWorkers
wg.Wait() // wait for all workers to exit
+ svr.pktMgr.close() // shuts down packetManager
// close any still-open files
for handle, file := range svr.openFiles {
@@ -321,6 +327,20 @@
return err // error from recvPacket
}
+// Wrap underlying connection methods to use packetManager
+func (svr Server) sendPacket(m encoding.BinaryMarshaler) error {
+ if pkt, ok := m.(responsePacket); ok {
+ svr.pktMgr.readyPacket(pkt)
+ } else {
+ return errors.Errorf("unexpected packet type %T", m)
+ }
+ return nil
+}
+
+func (svr Server) sendError(p ider, err error) error {
+ return svr.sendPacket(statusFromError(p, err))
+}
+
type ider interface {
id() uint32
}
