Integrating server routines

cmd/fred/fred.go

@@ -1,8 +1,6 @@
 package main
 
 import (
-	"log"
-	"net/http"
 	_ "net/http/pprof"
 	"os"
 	"os/signal"
@@ -161,9 +159,10 @@ func main() {
 	}
 
 	// Register RPC Commands and set server
-	rpc1 := new(cxauctionrpc.OpencxAuctionRPC)
-	rpc1.OffButton = make(chan bool, 1)
-	rpc1.Server = fredServer
+	var rpcListener *cxauctionrpc.AuctionRPCCaller
+	if rpcListener, err = cxauctionrpc.CreateRPCForServer(fredServer); err != nil {
+		logging.Fatalf("Error creating rpc caller for server: %s", err)
+	}
 
 	// SIGINT and SIGTERM and SIGQUIT handler for CTRL-c, KILL, CTRL-/, etc.
 	go func() {
@@ -177,29 +176,31 @@ func main() {
 			logging.Infof("Received %s signal, Stopping server gracefully...", signal.String())
 
 			// send off button to off button
-			rpc1.OffButton <- true
+			if err = rpcListener.KillServerNoWait(); err != nil {
+				logging.Fatalf("Error killing server: %s", err)
+			}
 
 			return
 		}
 	}()
 
-	doneChan := make(chan bool, 1)
 	if !conf.AuthenticatedRPC {
 		// this tells us when the rpclisten is done
 		logging.Infof(" === will start to listen on rpc ===")
-		go cxauctionrpc.RPCListenAsync(doneChan, rpc1, conf.Rpchost, conf.Rpcport)
+		if err = rpcListener.RPCListen(conf.Rpchost, conf.Rpcport); err != nil {
+			logging.Fatalf("Error listening for rpc for auction serer: %s", err)
+		}
 	} else {
 		privkey, _ := koblitz.PrivKeyFromBytes(koblitz.S256(), key[:])
 		// this tells us when the rpclisten is done
 		logging.Infof(" === will start to listen on noise-rpc ===")
-		go cxauctionrpc.NoiseListenAsync(doneChan, privkey, rpc1, conf.Rpchost, conf.Rpcport)
+		if err = rpcListener.NoiseListen(privkey, conf.Rpchost, conf.Rpcport); err != nil {
+			logging.Fatalf("Error listening for noise rpc for auction serer: %s", err)
+		}
 	}
 
-	go func() {
-		log.Println(http.ListenAndServe("localhost:6060", nil))
-	}()
-
-	<-doneChan
+	// wait until the listener dies
+	rpcListener.WaitUntilDead()
 
 	return
 }

cmd/ocx/auctioncmds.go

@@ -8,6 +8,7 @@ import (
 	"github.com/mit-dci/lit/lnutil"
 	"github.com/mit-dci/opencx/cxauctionrpc"
 	"github.com/mit-dci/opencx/logging"
+	"github.com/mit-dci/opencx/match"
 )
 
 var placeAuctionOrderCommand = &Command{
@@ -45,8 +46,14 @@ func (cl *ocxClient) AuctionOrderCommand(args []string) (err error) {
 		return
 	}
 
+	pairParam := new(match.Pair)
+	if err = pairParam.FromString(pair); err != nil {
+		err = fmt.Errorf("Error parsing pair, please enter something valid: %s", err)
+		return
+	}
+
 	var paramreply *cxauctionrpc.GetPublicParametersReply
-	if paramreply, err = cl.RPCClient.GetPublicParameters(); err != nil {
+	if paramreply, err = cl.RPCClient.GetPublicParameters(pairParam); err != nil {
 		err = fmt.Errorf("Error getting public parameters before placing auction order: %s", err)
 		return
 	}

cxauctionrpc/cxauctionrpc.go

@@ -1,9 +1,19 @@
 package cxauctionrpc
 
-import "github.com/mit-dci/opencx/cxauctionserver"
+import (
+	"net"
 
-// OpencxAuctionRPC is a listener for RPC commands
+	"github.com/mit-dci/opencx/cxauctionserver"
+)
+
+// AuctionRPCCaller is a listener for RPC commands
+type AuctionRPCCaller struct {
+	caller   *OpencxAuctionRPC
+	listener net.Listener
+	killers  []chan bool
+}
+
+// OpencxAuctionRPC is what is registered and called
 type OpencxAuctionRPC struct {
-	Server    *cxauctionserver.OpencxAuctionServer
-	OffButton chan bool
+	Server *cxauctionserver.OpencxAuctionServer
 }

cxauctionrpc/listener.go

@@ -6,93 +6,160 @@ import (
 	"net/rpc"
 
 	"github.com/mit-dci/lit/crypto/koblitz"
+	"github.com/mit-dci/opencx/cxauctionserver"
 	"github.com/mit-dci/opencx/cxnoise"
 	"github.com/mit-dci/opencx/logging"
 )
 
+func CreateRPCForServer(server *cxauctionserver.OpencxAuctionServer) (rpc1 *AuctionRPCCaller, err error) {
+	rpc1 = &AuctionRPCCaller{
+		caller: &OpencxAuctionRPC{
+			Server: server,
+		},
+	}
+	return
+}
+
 // NoiseListen is a synchronous version of RPCListenAsync
-func NoiseListen(rpc1 *OpencxAuctionRPC, privkey *koblitz.PrivateKey, host string, port uint16) {
+func (rpc1 *AuctionRPCCaller) NoiseListen(privkey *koblitz.PrivateKey, host string, port uint16) (err error) {
 
 	doneChan := make(chan bool, 1)
-	go NoiseListenAsync(doneChan, privkey, rpc1, host, port)
-	<-doneChan
+	errChan := make(chan error, 1)
+	go rpc1.NoiseListenAsync(doneChan, errChan, privkey, host, port)
+	select {
+	case err = <-errChan:
+	case <-doneChan:
+	}
+	close(doneChan)
+	close(errChan)
 
 	return
 }
 
 // NoiseListenAsync listens on socket host and port
-func NoiseListenAsync(doneChan chan bool, privkey *koblitz.PrivateKey, rpc1 *OpencxAuctionRPC, host string, port uint16) {
+func (rpc1 *AuctionRPCCaller) NoiseListenAsync(doneChan chan bool, errChan chan error, privkey *koblitz.PrivateKey, host string, port uint16) {
 	var err error
 
 	// Start noise rpc server (need to do this since the client is a rpc newclient)
 	noiseRPCServer := rpc.NewServer()
 
 	logging.Infof("Registering RPC API over Noise protocol ...")
 	// Register rpc
-	if err = noiseRPCServer.Register(rpc1); err != nil {
-		logging.Fatalf("Error registering RPC Interface:\n%s", err)
+	if err = noiseRPCServer.Register(rpc1.caller); err != nil {
+		errChan <- fmt.Errorf("Error registering RPC Interface:\n%s", err)
+		return
 	}
 
 	logging.Infof("Starting RPC Server over noise protocol")
 	// Start RPC Server
-	var listener net.Listener
-	if listener, err = cxnoise.NewListener(privkey, int(port)); err != nil {
-		logging.Fatal("listen error:", err)
+	if rpc1.listener, err = cxnoise.NewListener(privkey, int(port)); err != nil {
+		errChan <- fmt.Errorf("Error creating noise listener for NoiseListenAsync: %s", err)
+		return
 	}
-	logging.Infof("Running RPC-Noise server on %s\n", listener.Addr().String())
+	logging.Infof("Running RPC-Noise server on %s\n", rpc1.listener.Addr().String())
 
 	// We don't need to do anything fancy here either because the noise protocol
 	// is built in to the listener as well.
-	go noiseRPCServer.Accept(listener)
-	OffButtonCloseListener(rpc1, listener)
+	go noiseRPCServer.Accept(rpc1.listener)
 	doneChan <- true
 	return
 }
 
 // RPCListen is a synchronous version of RPCListenAsync
-func RPCListen(rpc1 *OpencxAuctionRPC, host string, port uint16) {
+func (rpc1 *AuctionRPCCaller) RPCListen(host string, port uint16) (err error) {
 
 	doneChan := make(chan bool, 1)
-	go RPCListenAsync(doneChan, rpc1, host, port)
-	<-doneChan
+	errChan := make(chan error, 1)
+	go rpc1.RPCListenAsync(doneChan, errChan, host, port)
+	select {
+	case err = <-errChan:
+	case <-doneChan:
+	}
+	close(doneChan)
+	close(errChan)
+
+	return
+}
+
+// KillServerNoWait kills the server, stops the clock, everything, doesn't
+func (rpc1 *AuctionRPCCaller) KillServerNoWait() (err error) {
+	if err = rpc1.caller.Server.StopClockDoNotWait(); err != nil {
+		err = fmt.Errorf("Error stopping clock, not waiting for results: %s", err)
+		return
+	}
+	if err = rpc1.Stop(); err != nil {
+		err = fmt.Errorf("Error stopping listener for KillServer: %s", err)
+		return
+	}
+	return
+}
 
+// KillServerWait kills the server, stops the clock, everything, but waits for stuff
+func (rpc1 *AuctionRPCCaller) KillServerWait() (err error) {
+	if err = rpc1.caller.Server.StopClockAndWait(); err != nil {
+		err = fmt.Errorf("Error stopping clock, waiting for results for KillServer: %s", err)
+		return
+	}
+	if err = rpc1.Stop(); err != nil {
+		err = fmt.Errorf("Error stopping listener for KillServer: %s", err)
+		return
+	}
 	return
 }
 
 // RPCListenAsync listens on socket host and port
-func RPCListenAsync(doneChan chan bool, rpc1 *OpencxAuctionRPC, host string, port uint16) {
+func (rpc1 *AuctionRPCCaller) RPCListenAsync(doneChan chan bool, errChan chan error, host string, port uint16) {
 	var err error
 
 	logging.Infof("Registering RPC API...")
 	// Register rpc
-	if err = rpc.Register(rpc1); err != nil {
-		logging.Fatalf("Error registering RPC Interface:\n%s", err)
+	if err = rpc.Register(rpc1.caller); err != nil {
+		errChan <- fmt.Errorf("Error registering RPC Interface:\n%s", err)
+		return
 	}
 
 	logging.Infof("Starting RPC Server")
 	// Start RPC Server
 	serverAddr := net.JoinHostPort(host, fmt.Sprintf("%d", port))
-	var listener net.Listener
-	if listener, err = net.Listen("tcp", serverAddr); err != nil {
-		logging.Fatal("listen error:", err)
+	if rpc1.listener, err = net.Listen("tcp", serverAddr); err != nil {
+		errChan <- fmt.Errorf("Error listening for RPCListenAsync: %s", err)
+		return
 	}
-	logging.Infof("Running RPC server on %s\n", listener.Addr().String())
+	logging.Infof("Running RPC server on %s\n", rpc1.listener.Addr().String())
 
-	go rpc.Accept(listener)
-
-	OffButtonCloseListener(rpc1, listener)
+	go rpc.Accept(rpc1.listener)
 	doneChan <- true
 	return
 }
 
-// OffButtonCloseListener waits for the off button to close the listener
-func OffButtonCloseListener(rpc1 *OpencxAuctionRPC, listener net.Listener) {
-	for {
-		<-rpc1.OffButton
-		logging.Infof("Got stop request, closing tcp listener")
-		if err := listener.Close(); err != nil {
-			logging.Errorf("Error closing listener: \n%s", err)
-		}
+// WaitUntilDead waits until the Stop() method is called
+func (rpc1 *AuctionRPCCaller) WaitUntilDead() {
+	dedchan := make(chan bool, 1)
+	rpc1.killers = append(rpc1.killers, dedchan)
+	<-dedchan
+	return
+}
+
+// Stop closes the RPC listener and notifies those from WaitUntilDead
+func (rpc1 *AuctionRPCCaller) Stop() (err error) {
+	if rpc1.listener == nil {
+		err = fmt.Errorf("Error, cannot stop a listener that doesn't exist")
+		return
+	}
+	logging.Infof("Stopping RPC!!")
+	if err = rpc1.listener.Close(); err != nil {
+		err = fmt.Errorf("Error closing listener: \n%s", err)
 		return
 	}
+	// kill the guy waiting
+	for _, killer := range rpc1.killers {
+		// send the signals, but even if they don't send, close the channel
+		select {
+		case killer <- true:
+			close(killer)
+		default:
+			close(killer)
+		}
+	}
+	return
 }