// TestSleepTest is an integration test for all of m_etcd's components.
//
TestSleepTest is an integration test for all of m_etcd's components.
// TestSleepTest is an integration test for all of m_etcd ' s components . //
func TestSleepTest(t *testing.T) {
	etcdv3c := testutil.NewEtcdV3Client(t)
	kvc := etcdv3.NewKV(etcdv3c)
	t.Parallel()
	const namespace = "/sleeptest-metafora"
	const sleepingtasks = "sleeping-task1"
	kvc.Delete(context.Background(), namespace, etcdv3.WithPrefix())
	holdtask := make(chan bool)
	h := func(task metafora.Task, cmds <-chan *statemachine.Message) *statemachine.Message {
		if task.ID() == sleepingtasks {
			sleeptil := 5 * time.Second
			nextstarttime := (time.Now().Add(sleeptil))
			t.Logf("sleeping task:%v sleepfor:%v", task, nextstarttime)
			<-holdtask
			return statemachine.SleepMessage(nextstarttime)
		}
		cmd := <-cmds
		t.Logf("non sleeping task:%v", task)
		return cmd
	}
	newC := func(name, ns string) *metafora.Consumer {
		conf := metcdv3.NewConfig(name, ns)
		coord, hf, bal := metcdv3.New(conf, etcdv3c, h)
		cons, err := metafora.NewConsumer(coord, hf, bal)
		if err != nil {
			t.Fatalf("Error creating consumer %s:%s: %v", ns, name, err)
		}
		go func() {
			cons.Run()
			t.Logf("Consumer:%s exited.", name)
		}()
		return cons
	}
	assertRunning := func(tid string, cons ...*metafora.Consumer) {
		found := false
		for _, c := range cons {
			tasks := c.Tasks()
			if len(tasks) > 0 && found {
				t.Fatal("Task already found running but another task is running on a different consumer")
			}
			if len(tasks) > 1 {
				t.Fatalf("Expected at most 1 task, but found: %d", len(tasks))
			}
			if len(tasks) == 1 && tasks[0].Task().ID() == tid {
				found = true
			}
		}
		if !found {
			t.Fatalf("Could not find task=%q", tid)
		}
	}
	cons1 := newC("node1", namespace)
	cons2 := newC("node2", namespace)
	cliA := metcdv3.NewClient(namespace, etcdv3c)
	if err := cliA.SubmitTask(metcdv3.DefaultTaskFunc(sleepingtasks, "")); err != nil {
		t.Fatalf("Error submitting task1 to a: %v", err)
	}
	time.Sleep(500 * time.Millisecond)
	assertRunning(sleepingtasks, cons1, cons2)
	holdtask <- true
	time.Sleep(500 * time.Millisecond)
	assertRunning(sleepingtasks, cons1, cons2) 
	wait1 := make(chan bool)
	go func() {
		defer close(wait1)
		cons1.Shutdown()
		cons2.Shutdown()
	}()
	timeout := time.NewTimer(5 * time.Second)
	select {
	case <-wait1:
	case <-timeout.C:
		t.Fatalf("failed waiting for shutdown")
	}
	for _, c := range []*metafora.Consumer{cons1, cons2} {
		tasks := c.Tasks()
		for _, work := range tasks {
			t.Fatalf("work id %v is still running", work)
		}
	}
}
func TestSleepTest ( t * testing . T ) { etcdv3c := testutil . NewEtcdV3Client ( t ) kvc := etcdv3 . NewKV ( etcdv3c ) t . Parallel ( ) const namespace = " " const sleepingtasks = " " kvc . Delete ( context . Background ( ) , namespace , etcdv3 . WithPrefix ( ) ) holdtask := make ( chan bool ) h := func ( task metafora . Task , cmds <- chan * statemachine . Message ) * statemachine . Message { if task . ID ( ) == sleepingtasks { sleeptil := 5 * time . Second nextstarttime := ( time . Now ( ) . Add ( sleeptil ) ) t . Logf ( " " , task , nextstarttime ) <- holdtask return statemachine . SleepMessage ( nextstarttime ) } cmd := <- cmds t . Logf ( " " , task ) return cmd } newC := func ( name , ns string ) * metafora . Consumer { conf := metcdv3 . NewConfig ( name , ns ) coord , hf , bal := metcdv3 . New ( conf , etcdv3c , h ) cons , err := metafora . NewConsumer ( coord , hf , bal ) if err != nil { t . Fatalf ( " " , ns , name , err ) } go func ( ) { cons . Run ( ) t . Logf ( " " , name ) } ( ) return cons } assertRunning := func ( tid string , cons ... * metafora . Consumer ) { found := false for _ , c := range cons { tasks := c . Tasks ( ) if len ( tasks ) > 0 && found { t . Fatal ( " " ) } if len ( tasks ) > 1 { t . Fatalf ( " " , len ( tasks ) ) } if len ( tasks ) == 1 && tasks [ 0 ] . Task ( ) . ID ( ) == tid { found = true } } if ! found { t . Fatalf ( " " , tid ) } } cons1 := newC ( " " , namespace ) cons2 := newC ( " " , namespace ) cliA := metcdv3 . NewClient ( namespace , etcdv3c ) if err := cliA . SubmitTask ( metcdv3 . DefaultTaskFunc ( sleepingtasks , " " ) ) ; err != nil { t . Fatalf ( " " , err ) } time . Sleep ( 500 * time . Millisecond ) assertRunning ( sleepingtasks , cons1 , cons2 ) holdtask <- true time . Sleep ( 500 * time . Millisecond ) assertRunning ( sleepingtasks , cons1 , cons2 ) wait1 := make ( chan bool ) go func ( ) { defer close ( wait1 ) cons1 . Shutdown ( ) cons2 . Shutdown ( ) } ( ) timeout := time . NewTimer ( 5 * time . Second ) select { case <- wait1 : case <- timeout . C : t . Fatalf ( " " ) } for _ , c := range [ ] * metafora . Consumer { cons1 , cons2 } { tasks := c . Tasks ( ) for _ , work := range tasks { t . Fatalf ( " " , work ) } } }
TestSleepTest is an integration test for all of m_etcd's components.
TestSleepTest is an integration test for all of m_etcd ' s components .
// Start 2 consumers // Create clients and start some tests // Give consumers a bit to pick up tasks // Give consumers a bit to pick up tasks // not sure if this should be true or false. // Shutdown //	make sure all tasks are released