example-scheduler: tidy up event handler builder
@@ -91,15 +91,12 @@ func Run(cfg Config) error {
// buildEventHandler generates and returns a handler to process events received from the subscription.
func buildEventHandler(state *internalState, frameworkIDStore store.Singleton) events.Handler {
logger := controller.LogEvents()
- return controller.LiftErrors().Handle(events.HandlerSet{
+ return controller.LiftErrors().DropOnError().Handle(events.HandlerSet{
scheduler.Event_FAILURE: logger.HandleF(failure),
- scheduler.Event_OFFERS: trackOffersReceived(state).AndThen().HandleF(
- func(e *scheduler.Event) error {
- if state.config.verbose {
- log.Println("received an OFFERS event")
- }
- return resourceOffers(state, e.GetOffers().GetOffers())
- }),
+ scheduler.Event_OFFERS: eventrules.Concat(
+ trackOffersReceived(state),
+ logger.If(state.config.verbose),
+ ).HandleF(resourceOffers(state)),
scheduler.Event_UPDATE: controller.AckStatusUpdates(state.cli).AndThen().HandleF(statusUpdate(state)),
scheduler.Event_SUBSCRIBED: eventrules.Rules{
logger,
@@ -140,10 +137,14 @@ func failure(e *scheduler.Event) error {
return nil
}
-func resourceOffers(state *internalState, offers []mesos.Offer) error {
- callOption := calls.RefuseSecondsWithJitter(state.random, state.config.maxRefuseSeconds)
- tasksLaunchedThisCycle := 0
- offersDeclined := 0
+func resourceOffers(state *internalState) events.HandlerFunc {
+ return func(e *scheduler.Event) error {
+ var (
+ offers = e.GetOffers().GetOffers()
+ callOption = calls.RefuseSecondsWithJitter(state.random, state.config.maxRefuseSeconds)
+ tasksLaunchedThisCycle = 0
+ offersDeclined = 0
+ )
for i := range offers {
var (
remaining = mesos.Resources(offers[i].Resources)
@@ -151,7 +152,8 @@ func resourceOffers(state *internalState, offers []mesos.Offer) error {
)
if state.config.verbose {
- log.Println("received offer id '" + offers[i].ID.Value + "' with resources " + remaining.String())
+ log.Println("received offer id '" + offers[i].ID.Value +
+ "' with resources " + remaining.String())
}
var wantsExecutorResources mesos.Resources
@@ -218,6 +220,7 @@ func resourceOffers(state *internalState, offers []mesos.Offer) error {
}
return nil
}
+}
func statusUpdate(state *internalState) events.HandlerFunc {
return func(e *scheduler.Event) error {
