func RemoteAuction(client yagnats.NATSClient, auctionRequest types.AuctionRequest) types.AuctionResult {
	guid := util.RandomGuid()
	payload, _ := json.Marshal(auctionRequest)

	c := make(chan []byte)
	client.Subscribe(guid, func(msg *yagnats.Message) {
		c <- msg.Payload
	})

	client.PublishWithReplyTo("diego.auction", guid, payload)

	var responsePayload []byte
	select {
	case responsePayload = <-c:
	case <-time.After(time.Minute):
		return types.AuctionResult{}
	}

	var auctionResult types.AuctionResult
	err := json.Unmarshal(responsePayload, &auctionResult)
	if err != nil {
		panic(err)
	}

	return auctionResult
}

func handleStaging(bbs bbs.StagerBBS, natsClient yagnats.NATSClient) {
	var task uint64

	natsClient.SubscribeWithQueue("stage", "stager", func(msg *yagnats.Message) {
		var message stagingMessage

		err := json.Unmarshal(msg.Payload, &message)
		if err != nil {
			logger.Fatal("staging-request.invalid", map[string]interface{}{
				"error":   err.Error(),
				"payload": string(msg.Payload),
			})
			return
		}

		for i := 0; i < message.Count; i++ {
			guid := atomic.AddUint64(&task, 1)

			task := &models.Task{
				Guid:     fmt.Sprintf("task-%d", guid),
				MemoryMB: message.MemoryMB,

				ReplyTo: msg.ReplyTo,
			}

			logger.Info("staging-request.desire", map[string]interface{}{
				"task": task,
			})

			go bbs.DesireTask(task)
		}
	})
}

func NewStartStopListener(messageBus yagnats.NATSClient, conf *config.Config) *StartStopListener {
	listener := &StartStopListener{
		messageBus: messageBus,
	}

	messageBus.Subscribe(conf.SenderNatsStartSubject, func(message *yagnats.Message) {
		startMessage, err := models.NewStartMessageFromJSON([]byte(message.Payload))
		if err != nil {
			panic(err)
		}
		listener.Starts = append(listener.Starts, startMessage)
	})

	messageBus.Subscribe(conf.SenderNatsStopSubject, func(message *yagnats.Message) {
		stopMessage, err := models.NewStopMessageFromJSON([]byte(message.Payload))
		if err != nil {
			panic(err)
		}
		listener.Stops = append(listener.Stops, stopMessage)
	})

	return listener
}

func Register(c *VcapComponent, mbusClient yagnats.NATSClient) {
	mbusClient.Subscribe("vcap.component.discover", func(msg *yagnats.Message) {
		Component.Uptime = Component.Start.Elapsed()
		b, e := json.Marshal(Component)
		if e != nil {
			log.Warnf(e.Error())
			return
		}

		mbusClient.Publish(msg.ReplyTo, b)
	})

	b, e := json.Marshal(Component)
	if e != nil {
		log.Fatal(e.Error())
		panic("Component's information should be correct")
	}

	mbusClient.Publish("vcap.component.announce", b)

	log.Infof("Component %s registered successfully", Component.Type)
}

func (c *VcapComponent) Register(mbusClient yagnats.NATSClient) error {
	mbusClient.Subscribe("vcap.component.discover", func(msg *yagnats.Message) {
		c.Uptime = c.StartTime.Elapsed()
		b, e := json.Marshal(c)
		if e != nil {
			log.Warnf(e.Error())
			return
		}

		mbusClient.Publish(msg.ReplyTo, b)
	})

	b, e := json.Marshal(c)
	if e != nil {
		log.Error(e.Error())
		return e
	}

	mbusClient.Publish("vcap.component.announce", b)

	log.Infof("Component %s registered successfully", c.Type)
	return nil
}

	"time"
)

var guids []string
var natsAddrs []string

var numReps int
var repResources int

var rules types.AuctionRules
var timeout time.Duration

var auctioneerMode string

// plumbing
var natsClient yagnats.NATSClient
var client types.TestRepPoolClient
var communicator types.AuctionCommunicator

func init() {
	flag.StringVar(&auctioneerMode, "auctioneerMode", "inprocess", "one of inprocess, remote")

	flag.IntVar(&(auctioneer.DefaultRules.MaxRounds), "maxRounds", auctioneer.DefaultRules.MaxRounds, "the maximum number of rounds per auction")
	flag.IntVar(&(auctioneer.DefaultRules.MaxBiddingPool), "maxBiddingPool", auctioneer.DefaultRules.MaxBiddingPool, "the maximum number of participants in the pool")
	flag.IntVar(&(auctioneer.DefaultRules.MaxConcurrent), "maxConcurrent", auctioneer.DefaultRules.MaxConcurrent, "the maximum number of concurrent auctions to run")
	flag.BoolVar(&(auctioneer.DefaultRules.RepickEveryRound), "repickEveryRound", auctioneer.DefaultRules.RepickEveryRound, "whether to repick every round")
}

func TestAuction(t *testing.T) {
	RegisterFailHandler(Fail)
	RunSpecs(t, "Auction Suite")

func runSimulation(natsClient yagnats.NATSClient) {
	simulationLock = &sync.Mutex{}
	simulationWait = &sync.WaitGroup{}
	taskTracker = map[string]*taskData{}

	msg := stagingMessage{
		Count:    nTasks,
		MemoryMB: taskMemory,
	}

	payload, err := json.Marshal(msg)
	if err != nil {
		panic(err)
	}

	t := time.Now()

	logger.Info("simulation.start", nTasks)

	simulationWait.Add(nTasks)

	_, err = natsClient.Subscribe("info.stager.*.staging-request.desire", func(msg *yagnats.Message) {
		var desiredLog struct {
			Timestamp time.Time   `json:"_timestamp"`
			Task      models.Task `json:"task"`
		}

		err := json.Unmarshal(msg.Payload, &desiredLog)
		if err != nil {
			panic(err)
		}

		registerDesired(desiredLog.Task.Guid, desiredLog.Timestamp)
	})

	_, err = natsClient.Subscribe("error.>", func(msg *yagnats.Message) {
		var errorLog struct {
			Timestamp time.Time `json:"_timestamp"`
			Error     string    `json:"error"`
		}

		err := json.Unmarshal(msg.Payload, &errorLog)
		if err != nil {
			panic(err)
		}

		registerError(msg.Subject+": "+errorLog.Error, errorLog.Timestamp)
	})

	_, err = natsClient.Subscribe("fatal.>", func(msg *yagnats.Message) {
		var errorLog struct {
			Timestamp time.Time `json:"_timestamp"`
			Error     string    `json:"error"`
		}

		err := json.Unmarshal(msg.Payload, &errorLog)
		if err != nil {
			panic(err)
		}

		registerError(msg.Subject+": "+errorLog.Error, errorLog.Timestamp)
	})

	executorIndexes := map[string]int{}

	_, err = natsClient.Subscribe("completed-task", func(msg *yagnats.Message) {
		defer func() {
			e := recover()
			if e != nil {
				logger.Error("RECOVERED PANIC:", e)
			}
		}()

		var task *models.Task

		err := json.Unmarshal(msg.Payload, &task)
		if err != nil {
			panic(err)
		}

		simulationLock.Lock()

		index, ok := executorIndexes[task.ExecutorID]
		if !ok {
			index = len(executorIndexes) + 1
			executorIndexes[task.ExecutorID] = index
		}

		data, ok := taskTracker[task.Guid]
		if !ok {
			logger.Error("uknown.runonce.completed", task.Guid, "executor", task.ExecutorID)
			simulationLock.Unlock()
			return
		}

		data.CompletionTime = float64(time.Now().UnixNano()) / 1e9
		logger.Info("runonce.completed", task.Guid, "executor", task.ExecutorID, "duration", data.CompletionTime-data.DesiredTime)
		data.ExecutorIndex = index
		data.NumCompletions++

//.........这里部分代码省略.........