Make ServiceQueue a separate config param.

Also, make clientName strings into constants.

cmd/activity-monitor/main.go

@@ -115,8 +115,7 @@ func main() {
 	app.Action = func(c cmd.Config, stats statsd.Statter, auditlogger *blog.AuditLogger) {
 		go cmd.DebugServer(c.ActivityMonitor.DebugAddr)
 
-		amqpConf := c.ActivityMonitor.AMQP
+		ch, err := rpc.AmqpChannel(c.ActivityMonitor.AMQP)
-		ch, err := rpc.AmqpChannel(amqpConf)
 
 		cmd.FailOnError(err, "Could not connect to AMQP")
 

cmd/admin-revoker/main.go

@@ -36,6 +36,8 @@ func loadConfig(c *cli.Context) (config cmd.Config, err error) {
 	return
 }
 
+const clientName = "AdminRevoker"
+
 func setupContext(context *cli.Context) (rpc.RegistrationAuthorityClient, *blog.AuditLogger, *gorp.DbMap, rpc.StorageAuthorityClient) {
 	c, err := loadConfig(context)
 	cmd.FailOnError(err, "Failed to load Boulder configuration")
@@ -43,7 +45,6 @@ func setupContext(context *cli.Context) (rpc.RegistrationAuthorityClient, *blog.
 	stats, auditlogger := cmd.StatsAndLogging(c.Statsd, c.Syslog)
 
 	amqpConf := c.Revoker.AMQP
-	clientName := "AdminRevoker"
 	rac, err := rpc.NewRegistrationAuthorityClient(clientName, amqpConf, stats)
 	cmd.FailOnError(err, "Unable to create CA client")
 

cmd/boulder-ca/main.go

@@ -16,6 +16,8 @@ import (
 	"github.com/letsencrypt/boulder/sa"
 )
 
+const clientName = "CA"
+
 func main() {
 	app := cmd.NewAppShell("boulder-ca", "Handles issuance operations")
 	app.Action = func(c cmd.Config, stats statsd.Statter, auditlogger *blog.AuditLogger) {
@@ -42,14 +44,13 @@ func main() {
 		go cmd.ProfileCmd("CA", stats)
 
 		amqpConf := c.CA.AMQP
-		clientName := "CA"
 		cai.SA, err = rpc.NewStorageAuthorityClient(clientName, amqpConf, stats)
 		cmd.FailOnError(err, "Failed to create SA client")
 
 		cai.Publisher, err = rpc.NewPublisherClient(clientName, amqpConf, stats)
 		cmd.FailOnError(err, "Failed to create Publisher client")
 
-		cas, err := rpc.NewAmqpRPCServer(amqpConf, amqpConf.CA, c.CA.MaxConcurrentRPCServerRequests, stats)
+		cas, err := rpc.NewAmqpRPCServer(amqpConf, c.CA.MaxConcurrentRPCServerRequests, stats)
 		cmd.FailOnError(err, "Unable to create CA RPC server")
 		rpc.NewCertificateAuthorityServer(cas, cai)
 

cmd/boulder-publisher/main.go

@@ -14,6 +14,8 @@ import (
 	"github.com/letsencrypt/boulder/rpc"
 )
 
+const clientName = "Publisher"
+
 func main() {
 	app := cmd.NewAppShell("boulder-publisher", "Submits issued certificates to CT logs")
 	app.Action = func(c cmd.Config, stats statsd.Statter, auditlogger *blog.AuditLogger) {
@@ -23,13 +25,11 @@ func main() {
 		go cmd.DebugServer(c.Publisher.DebugAddr)
 		go cmd.ProfileCmd("Publisher", stats)
 
-		pubConf := c.Publisher
+		amqpConf := c.Publisher.AMQP
-		amqpConf := pubConf.AMQP
-		clientName := "Publisher"
 		pubi.SA, err = rpc.NewStorageAuthorityClient(clientName, amqpConf, stats)
 		cmd.FailOnError(err, "Unable to create SA client")
 
-		pubs, err := rpc.NewAmqpRPCServer(amqpConf, amqpConf.Publisher, pubConf.MaxConcurrentRPCServerRequests, stats)
+		pubs, err := rpc.NewAmqpRPCServer(amqpConf, c.Publisher.MaxConcurrentRPCServerRequests, stats)
 		cmd.FailOnError(err, "Unable to create Publisher RPC server")
 		rpc.NewPublisherServer(pubs, &pubi)
 

cmd/boulder-ra/main.go

@@ -20,6 +20,8 @@ import (
 	"github.com/letsencrypt/boulder/rpc"
 )
 
+const clientName = "RA"
+
 func main() {
 	app := cmd.NewAppShell("boulder-ra", "Handles service orchestration")
 	app.Action = func(c cmd.Config, stats statsd.Statter, auditlogger *blog.AuditLogger) {
@@ -40,7 +42,6 @@ func main() {
 		go cmd.ProfileCmd("RA", stats)
 
 		amqpConf := c.RA.AMQP
-		clientName := "RA"
 		vac, err := rpc.NewValidationAuthorityClient(clientName, amqpConf, stats)
 		cmd.FailOnError(err, "Unable to create VA client")
 
@@ -70,7 +71,7 @@ func main() {
 		rai.CA = cac
 		rai.SA = sac
 
-		ras, err := rpc.NewAmqpRPCServer(amqpConf, amqpConf.RA, c.RA.MaxConcurrentRPCServerRequests, stats)
+		ras, err := rpc.NewAmqpRPCServer(amqpConf, c.RA.MaxConcurrentRPCServerRequests, stats)
 		cmd.FailOnError(err, "Unable to create RA RPC server")
 		rpc.NewRegistrationAuthorityServer(ras, rai)
 

cmd/boulder-sa/main.go

@@ -30,7 +30,7 @@ func main() {
 		go cmd.ProfileCmd("SA", stats)
 
 		amqpConf := saConf.AMQP
-		sas, err := rpc.NewAmqpRPCServer(amqpConf, amqpConf.SA, c.SA.MaxConcurrentRPCServerRequests, stats)
+		sas, err := rpc.NewAmqpRPCServer(amqpConf, c.SA.MaxConcurrentRPCServerRequests, stats)
 		cmd.FailOnError(err, "Unable to create SA RPC server")
 		rpc.NewStorageAuthorityServer(sas, sai)
 

cmd/boulder-va/main.go

@@ -18,6 +18,8 @@ import (
 	"github.com/letsencrypt/boulder/va"
 )
 
+const clientName = "VA"
+
 func main() {
 	app := cmd.NewAppShell("boulder-va", "Handles challenge validation")
 	app.Action = func(c cmd.Config, stats statsd.Statter, auditlogger *blog.AuditLogger) {
@@ -51,13 +53,12 @@ func main() {
 		vai.UserAgent = c.VA.UserAgent
 
 		amqpConf := c.VA.AMQP
-		clientName := "VA"
 		rac, err := rpc.NewRegistrationAuthorityClient(clientName, amqpConf, stats)
 		cmd.FailOnError(err, "Unable to create RA client")
 
 		vai.RA = rac
 
-		vas, err := rpc.NewAmqpRPCServer(amqpConf, amqpConf.VA, c.VA.MaxConcurrentRPCServerRequests, stats)
+		vas, err := rpc.NewAmqpRPCServer(amqpConf, c.VA.MaxConcurrentRPCServerRequests, stats)
 		cmd.FailOnError(err, "Unable to create VA RPC server")
 		rpc.NewValidationAuthorityServer(vas, vai)
 

cmd/boulder-wfe/main.go

@@ -22,9 +22,10 @@ import (
 	"github.com/letsencrypt/boulder/wfe"
 )
 
+const clientName = "WFE"
+
 func setupWFE(c cmd.Config, logger *blog.AuditLogger, stats statsd.Statter) (*rpc.RegistrationAuthorityClient, *rpc.StorageAuthorityClient) {
 	amqpConf := c.WFE.AMQP
-	clientName := "WFE"
 	rac, err := rpc.NewRegistrationAuthorityClient(clientName, amqpConf, stats)
 	cmd.FailOnError(err, "Unable to create RA client")
 

cmd/config.go

@@ -183,14 +183,17 @@ type ServiceConfig struct {
 // AMQPConfig describes how to connect to AMQP, and how to speak to each of the
 // RPC services we offer via AMQP.
 type AMQPConfig struct {
-	Server            string
+	Server    string
-	Insecure          bool
+	Insecure  bool
-	RA                *RPCServerConfig
+	RA        *RPCServerConfig
-	VA                *RPCServerConfig
+	VA        *RPCServerConfig
-	SA                *RPCServerConfig
+	SA        *RPCServerConfig
-	CA                *RPCServerConfig
+	CA        *RPCServerConfig
-	Publisher         *RPCServerConfig
+	Publisher *RPCServerConfig
-	TLS               *TLSConfig
+	TLS       *TLSConfig
+	// Queue name on which to listen, if this is an RPC service (vs acting only as
+	// an RPC client).
+	ServiceQueue      string
 	ReconnectTimeouts struct {
 		Base ConfigDuration
 		Max  ConfigDuration

cmd/expiration-mailer/main.go

@@ -208,6 +208,8 @@ func (ds durationSlice) Swap(a, b int) {
 	ds[a], ds[b] = ds[b], ds[a]
 }
 
+const clientName = "ExpirationMailer"
+
 func main() {
 	app := cmd.NewAppShell("expiration-mailer", "Sends certificate expiration emails")
 
@@ -233,7 +235,6 @@ func main() {
 		cmd.FailOnError(err, "Could not connect to database")
 
 		amqpConf := c.SA.AMQP
-		clientName := "ExpirationMailer"
 		sac, err := rpc.NewStorageAuthorityClient(clientName, amqpConf, stats)
 		cmd.FailOnError(err, "Failed to create SA client")
 

cmd/ocsp-updater/main.go

@@ -531,13 +531,14 @@ func (l *looper) loop() error {
 	}
 }
 
+const clientName = "OCSP"
+
 func setupClients(c cmd.OCSPUpdaterConfig, stats statsd.Statter) (
 	core.CertificateAuthority,
 	core.Publisher,
 	core.StorageAuthority,
 ) {
 	amqpConf := c.AMQP
-	clientName := "OCSP"
 	cac, err := rpc.NewCertificateAuthorityClient(clientName, amqpConf, stats)
 	cmd.FailOnError(err, "Unable to create CA client")
 

rpc/amqp-rpc.go

@@ -177,7 +177,7 @@ type AmqpRPCServer struct {
 
 // NewAmqpRPCServer creates a new RPC server for the given queue and will begin
 // consuming requests from the queue. To start the server you must call Start().
-func NewAmqpRPCServer(amqpConf *cmd.AMQPConfig, rpcConf *cmd.RPCServerConfig, maxConcurrentRPCServerRequests int64, stats statsd.Statter) (*AmqpRPCServer, error) {
+func NewAmqpRPCServer(amqpConf *cmd.AMQPConfig, maxConcurrentRPCServerRequests int64, stats statsd.Statter) (*AmqpRPCServer, error) {
 	log := blog.GetAuditLogger()
 
 	reconnectBase := amqpConf.ReconnectTimeouts.Base.Duration
@@ -190,8 +190,8 @@ func NewAmqpRPCServer(amqpConf *cmd.AMQPConfig, rpcConf *cmd.RPCServerConfig, ma
 	}
 
 	return &AmqpRPCServer{
-		serverQueue:                    rpcConf.Server,
+		serverQueue:                    amqpConf.ServiceQueue,
-		connection:                     newAMQPConnector(rpcConf.Server, reconnectBase, reconnectMax),
+		connection:                     newAMQPConnector(amqpConf.ServiceQueue, reconnectBase, reconnectMax),
 		log:                            log,
 		dispatchTable:                  make(map[string]func([]byte) ([]byte, error)),
 		maxConcurrentRPCServerRequests: maxConcurrentRPCServerRequests,

test/boulder-config.json

@@ -98,7 +98,7 @@
     "amqp": {
       "server": "amqp://guest:guest@localhost:5673",
       "insecure": true,
-      "CA": { "server": "CA.server" },
+      "serviceQueue": "CA.server",
       "SA": {
         "server": "SA.server",
         "rpcTimeout": "1s"
@@ -129,7 +129,7 @@
     "amqp": {
       "server": "amqp://guest:guest@localhost:5673",
       "insecure": true,
-      "RA": { "server": "RA.server" },
+      "serviceQueue": "RA.server",
       "VA": {
         "server": "VA.server",
         "rpcTimeout": "60s"
@@ -153,7 +153,7 @@
     "amqp": {
       "server": "amqp://guest:guest@localhost:5673",
       "insecure": true,
-      "SA": { "server": "SA.server" }
+      "serviceQueue": "SA.server"
     }
   },
 
@@ -169,7 +169,7 @@
     "amqp": {
       "server": "amqp://guest:guest@localhost:5673",
       "insecure": true,
-      "VA": { "server": "VA.server" },
+      "serviceQueue": "VA.server",
       "RA": {
         "server": "RA.server",
         "rpcTimeout": "1s"
@@ -267,7 +267,7 @@
     "amqp": {
       "server": "amqp://guest:guest@localhost:5673",
       "insecure": true,
-      "Publisher": { "server": "Publisher.server" },
+      "serviceQueue": "Publisher.server",
       "SA": {
         "server": "SA.server",
         "rpcTimeout": "1s"