How to use Timeout method of got Package

...21	"github.com/nats-io/nats.go"22)23func TestContextRequestWithNilConnection(t *testing.T) {24	var nc *nats.Conn25	ctx, cancelCB := context.WithTimeout(context.Background(), 100*time.Millisecond)26	defer cancelCB() // should always be called, not discarded, to prevent context leak27	_, err := nc.RequestWithContext(ctx, "fast", []byte(""))28	if err == nil {29		t.Fatal("Expected request with context and nil connection to fail")30	}31	if err != nats.ErrInvalidConnection {32		t.Fatalf("Expected nats.ErrInvalidConnection, got %v\n", err)33	}34}35func testContextRequestWithTimeout(t *testing.T, nc *nats.Conn) {36	nc.Subscribe("slow", func(m *nats.Msg) {37		// Simulates latency into the client so that timeout is hit.38		time.Sleep(200 * time.Millisecond)39		nc.Publish(m.Reply, []byte("NG"))40	})41	nc.Subscribe("fast", func(m *nats.Msg) {42		nc.Publish(m.Reply, []byte("OK"))43	})44	ctx, cancelCB := context.WithTimeout(context.Background(), 100*time.Millisecond)45	defer cancelCB() // should always be called, not discarded, to prevent context leak46	// Fast request should not fail at this point.47	resp, err := nc.RequestWithContext(ctx, "fast", []byte(""))48	if err != nil {49		t.Fatalf("Expected request with context to not fail on fast response: %s", err)50	}51	got := string(resp.Data)52	expected := "OK"53	if got != expected {54		t.Errorf("Expected to receive %s, got: %s", expected, got)55	}56	// Slow request hits timeout so expected to fail.57	_, err = nc.RequestWithContext(ctx, "slow", []byte("world"))58	if err == nil {59		t.Fatal("Expected request with timeout context to fail")60	}61	// Reported error is "context deadline exceeded" from Context package,62	// which implements net.Error interface.63	type timeoutError interface {64		Timeout() bool65	}66	timeoutErr, ok := err.(timeoutError)67	if !ok || !timeoutErr.Timeout() {68		t.Error("Expected to have a timeout error")69	}70	expected = `context deadline exceeded`71	if !strings.Contains(err.Error(), expected) {72		t.Errorf("Expected %q error, got: %q", expected, err.Error())73	}74	// 2nd request should fail again even if they would be fast because context75	// has already timed out.76	_, err = nc.RequestWithContext(ctx, "fast", []byte("world"))77	if err == nil {78		t.Fatal("Expected request with context to fail")79	}80}81func TestContextRequestWithTimeout(t *testing.T) {82	s := RunDefaultServer()83	defer s.Shutdown()84	nc := NewDefaultConnection(t)85	defer nc.Close()86	testContextRequestWithTimeout(t, nc)87}88func TestOldContextRequestWithTimeout(t *testing.T) {89	s := RunDefaultServer()90	defer s.Shutdown()91	nc, err := nats.Connect(nats.DefaultURL, nats.UseOldRequestStyle())92	if err != nil {93		t.Fatalf("Failed to connect: %v", err)94	}95	defer nc.Close()96	testContextRequestWithTimeout(t, nc)97}98func testContextRequestWithTimeoutCanceled(t *testing.T, nc *nats.Conn) {99	ctx, cancelCB := context.WithTimeout(context.Background(), 100*time.Millisecond)100	defer cancelCB()101	nc.Subscribe("fast", func(m *nats.Msg) {102		nc.Publish(m.Reply, []byte("OK"))103	})104	// Fast request should not fail105	resp, err := nc.RequestWithContext(ctx, "fast", []byte(""))106	if err != nil {107		t.Fatalf("Expected request with context to not fail on fast response: %s", err)108	}109	got := string(resp.Data)110	expected := "OK"111	if got != expected {112		t.Errorf("Expected to receive %s, got: %s", expected, got)113	}114	// Cancel the context already so that rest of requests fail.115	cancelCB()116	// Context is already canceled so requests should immediately fail.117	_, err = nc.RequestWithContext(ctx, "fast", []byte("world"))118	if err == nil {119		t.Fatal("Expected request with timeout context to fail")120	}121	// Reported error is "context canceled" from Context package,122	// which is not a timeout error.123	type timeoutError interface {124		Timeout() bool125	}126	if _, ok := err.(timeoutError); ok {127		t.Errorf("Expected to not have a timeout error")128	}129	expected = `context canceled`130	if !strings.Contains(err.Error(), expected) {131		t.Errorf("Expected %q error, got: %q", expected, err.Error())132	}133	// 2nd request should fail again even if fast because context has already been canceled134	_, err = nc.RequestWithContext(ctx, "fast", []byte("world"))135	if err == nil {136		t.Fatal("Expected request with context to fail")137	}138}139func TestContextRequestWithTimeoutCanceled(t *testing.T) {140	s := RunDefaultServer()141	defer s.Shutdown()142	nc := NewDefaultConnection(t)143	defer nc.Close()144	testContextRequestWithTimeoutCanceled(t, nc)145}146func TestOldContextRequestWithTimeoutCanceled(t *testing.T) {147	s := RunDefaultServer()148	defer s.Shutdown()149	nc, err := nats.Connect(nats.DefaultURL, nats.UseOldRequestStyle())150	if err != nil {151		t.Fatalf("Failed to connect: %v", err)152	}153	defer nc.Close()154	testContextRequestWithTimeoutCanceled(t, nc)155}156func testContextRequestWithCancel(t *testing.T, nc *nats.Conn) {157	ctx, cancelCB := context.WithCancel(context.Background())158	defer cancelCB() // should always be called, not discarded, to prevent context leak159	// timer which cancels the context though can also be arbitrarily extended160	expirationTimer := time.AfterFunc(100*time.Millisecond, func() {161		cancelCB()162	})163	nc.Subscribe("slow", func(m *nats.Msg) {164		// simulates latency into the client so that timeout is hit.165		time.Sleep(40 * time.Millisecond)166		nc.Publish(m.Reply, []byte("OK"))167	})168	nc.Subscribe("slower", func(m *nats.Msg) {169		// we know this request will take longer so extend the timeout170		expirationTimer.Reset(100 * time.Millisecond)171		// slower reply which would have hit original timeout172		time.Sleep(90 * time.Millisecond)173		nc.Publish(m.Reply, []byte("Also OK"))174	})175	for i := 0; i < 2; i++ {176		resp, err := nc.RequestWithContext(ctx, "slow", []byte(""))177		if err != nil {178			t.Fatalf("Expected request with context to not fail: %s", err)179		}180		got := string(resp.Data)181		expected := "OK"182		if got != expected {183			t.Errorf("Expected to receive %s, got: %s", expected, got)184		}185	}186	// A third request with latency would make the context187	// get canceled, but these reset the timer so deadline188	// gets extended:189	for i := 0; i < 10; i++ {190		resp, err := nc.RequestWithContext(ctx, "slower", []byte(""))191		if err != nil {192			t.Fatalf("Expected request with context to not fail: %s", err)193		}194		got := string(resp.Data)195		expected := "Also OK"196		if got != expected {197			t.Errorf("Expected to receive %s, got: %s", expected, got)198		}199	}200	// One more slow request will expire the timer and cause an error...201	_, err := nc.RequestWithContext(ctx, "slow", []byte(""))202	if err == nil {203		t.Fatal("Expected request with cancellation context to fail")204	}205	// ...though reported error is "context canceled" from Context package,206	// which is not a timeout error.207	type timeoutError interface {208		Timeout() bool209	}210	if _, ok := err.(timeoutError); ok {211		t.Errorf("Expected to not have a timeout error")212	}213	expected := `context canceled`214	if !strings.Contains(err.Error(), expected) {215		t.Errorf("Expected %q error, got: %q", expected, err.Error())216	}217}218func TestContextOldRequestClosed(t *testing.T) {219	s := RunDefaultServer()220	defer s.Shutdown()221	nc, err := nats.Connect(nats.DefaultURL, nats.UseOldRequestStyle())222	if err != nil {223		t.Fatalf("Failed to connect: %v", err)224	}225	defer nc.Close()226	ctx, cancelCB := context.WithTimeout(context.Background(), time.Second)227	defer cancelCB() // should always be called, not discarded, to prevent context leak228	errCh := make(chan error, 1)229	start := time.Now()230	go func() {231		_, err = nc.RequestWithContext(ctx, "checkClose", []byte("should be kicked out on close"))232		errCh <- err233	}()234	time.Sleep(100 * time.Millisecond)235	nc.Close()236	if e := <-errCh; e != nats.ErrConnectionClosed {237		t.Fatalf("Unexpected error: %v", err)238	}239	if dur := time.Since(start); dur >= time.Second {240		t.Fatalf("Request took too long to bail out: %v", dur)241	}242}243func TestContextRequestWithCancel(t *testing.T) {244	s := RunDefaultServer()245	defer s.Shutdown()246	nc := NewDefaultConnection(t)247	defer nc.Close()248	testContextRequestWithCancel(t, nc)249}250func TestOldContextRequestWithCancel(t *testing.T) {251	s := RunDefaultServer()252	defer s.Shutdown()253	nc, err := nats.Connect(nats.DefaultURL, nats.UseOldRequestStyle())254	if err != nil {255		t.Fatalf("Failed to connect: %v", err)256	}257	defer nc.Close()258	testContextRequestWithCancel(t, nc)259}260func testContextRequestWithDeadline(t *testing.T, nc *nats.Conn) {261	deadline := time.Now().Add(100 * time.Millisecond)262	ctx, cancelCB := context.WithDeadline(context.Background(), deadline)263	defer cancelCB() // should always be called, not discarded, to prevent context leak264	nc.Subscribe("slow", func(m *nats.Msg) {265		// simulates latency into the client so that timeout is hit.266		time.Sleep(40 * time.Millisecond)267		nc.Publish(m.Reply, []byte("OK"))268	})269	for i := 0; i < 2; i++ {270		resp, err := nc.RequestWithContext(ctx, "slow", []byte(""))271		if err != nil {272			t.Fatalf("Expected request with context to not fail: %s", err)273		}274		got := string(resp.Data)275		expected := "OK"276		if got != expected {277			t.Errorf("Expected to receive %s, got: %s", expected, got)278		}279	}280	// A third request with latency would make the context281	// reach the deadline.282	_, err := nc.RequestWithContext(ctx, "slow", []byte(""))283	if err == nil {284		t.Fatal("Expected request with context to reach deadline")285	}286	// Reported error is "context deadline exceeded" from Context package,287	// which implements net.Error Timeout interface.288	type timeoutError interface {289		Timeout() bool290	}291	timeoutErr, ok := err.(timeoutError)292	if !ok || !timeoutErr.Timeout() {293		t.Errorf("Expected to have a timeout error")294	}295	expected := `context deadline exceeded`296	if !strings.Contains(err.Error(), expected) {297		t.Errorf("Expected %q error, got: %q", expected, err.Error())298	}299}300func TestContextRequestWithDeadline(t *testing.T) {301	s := RunDefaultServer()302	defer s.Shutdown()303	nc := NewDefaultConnection(t)304	defer nc.Close()305	testContextRequestWithDeadline(t, nc)306}307func TestOldContextRequestWithDeadline(t *testing.T) {308	s := RunDefaultServer()309	defer s.Shutdown()310	nc, err := nats.Connect(nats.DefaultURL, nats.UseOldRequestStyle())311	if err != nil {312		t.Fatalf("Failed to connect: %v", err)313	}314	defer nc.Close()315	testContextRequestWithDeadline(t, nc)316}317func TestContextSubNextMsgWithTimeout(t *testing.T) {318	s := RunDefaultServer()319	defer s.Shutdown()320	nc := NewDefaultConnection(t)321	defer nc.Close()322	ctx, cancelCB := context.WithTimeout(context.Background(), 100*time.Millisecond)323	defer cancelCB() // should always be called, not discarded, to prevent context leak324	sub, err := nc.SubscribeSync("slow")325	if err != nil {326		t.Fatalf("Expected to be able to subscribe: %s", err)327	}328	for i := 0; i < 2; i++ {329		err := nc.Publish("slow", []byte("OK"))330		if err != nil {331			t.Fatalf("Expected publish to not fail: %s", err)332		}333		// Enough time to get a couple of messages334		time.Sleep(40 * time.Millisecond)335		msg, err := sub.NextMsgWithContext(ctx)336		if err != nil {337			t.Fatalf("Expected to receive message: %s", err)338		}339		got := string(msg.Data)340		expected := "OK"341		if got != expected {342			t.Errorf("Expected to receive %s, got: %s", expected, got)343		}344	}345	// Third message will fail because the context will be canceled by now346	_, err = sub.NextMsgWithContext(ctx)347	if err == nil {348		t.Fatal("Expected to fail receiving a message")349	}350	// Reported error is "context deadline exceeded" from Context package,351	// which implements net.Error Timeout interface.352	type timeoutError interface {353		Timeout() bool354	}355	timeoutErr, ok := err.(timeoutError)356	if !ok || !timeoutErr.Timeout() {357		t.Errorf("Expected to have a timeout error")358	}359	expected := `context deadline exceeded`360	if !strings.Contains(err.Error(), expected) {361		t.Errorf("Expected %q error, got: %q", expected, err.Error())362	}363}364func TestContextSubNextMsgWithTimeoutCanceled(t *testing.T) {365	s := RunDefaultServer()366	defer s.Shutdown()367	nc := NewDefaultConnection(t)368	defer nc.Close()369	ctx, cancelCB := context.WithTimeout(context.Background(), 100*time.Millisecond)370	defer cancelCB() // should always be called, not discarded, to prevent context leak371	sub, err := nc.SubscribeSync("fast")372	if err != nil {373		t.Fatalf("Expected to be able to subscribe: %s", err)374	}375	for i := 0; i < 2; i++ {376		err := nc.Publish("fast", []byte("OK"))377		if err != nil {378			t.Fatalf("Expected publish to not fail: %s", err)379		}380		// Enough time to get a couple of messages381		time.Sleep(40 * time.Millisecond)382		msg, err := sub.NextMsgWithContext(ctx)383		if err != nil {384			t.Fatalf("Expected to receive message: %s", err)385		}386		got := string(msg.Data)387		expected := "OK"388		if got != expected {389			t.Errorf("Expected to receive %s, got: %s", expected, got)390		}391	}392	// Cancel the context already so that rest of NextMsg calls fail.393	cancelCB()394	_, err = sub.NextMsgWithContext(ctx)395	if err == nil {396		t.Fatal("Expected request with timeout context to fail")397	}398	// Reported error is "context canceled" from Context package,399	// which is not a timeout error.400	type timeoutError interface {401		Timeout() bool402	}403	if _, ok := err.(timeoutError); ok {404		t.Errorf("Expected to not have a timeout error")405	}406	expected := `context canceled`407	if !strings.Contains(err.Error(), expected) {408		t.Errorf("Expected %q error, got: %q", expected, err.Error())409	}410}411func TestContextSubNextMsgWithCancel(t *testing.T) {412	s := RunDefaultServer()413	defer s.Shutdown()414	nc := NewDefaultConnection(t)415	defer nc.Close()416	ctx, cancelCB := context.WithCancel(context.Background())417	defer cancelCB() // should always be called, not discarded, to prevent context leak418	// timer which cancels the context though can also be arbitrarily extended419	time.AfterFunc(100*time.Millisecond, func() {420		cancelCB()421	})422	sub1, err := nc.SubscribeSync("foo")423	if err != nil {424		t.Fatalf("Expected to be able to subscribe: %s", err)425	}426	sub2, err := nc.SubscribeSync("bar")427	if err != nil {428		t.Fatalf("Expected to be able to subscribe: %s", err)429	}430	for i := 0; i < 2; i++ {431		err := nc.Publish("foo", []byte("OK"))432		if err != nil {433			t.Fatalf("Expected publish to not fail: %s", err)434		}435		resp, err := sub1.NextMsgWithContext(ctx)436		if err != nil {437			t.Fatalf("Expected request with context to not fail: %s", err)438		}439		got := string(resp.Data)440		expected := "OK"441		if got != expected {442			t.Errorf("Expected to receive %s, got: %s", expected, got)443		}444	}445	err = nc.Publish("bar", []byte("Also OK"))446	if err != nil {447		t.Fatalf("Expected publish to not fail: %s", err)448	}449	resp, err := sub2.NextMsgWithContext(ctx)450	if err != nil {451		t.Fatalf("Expected request with context to not fail: %s", err)452	}453	got := string(resp.Data)454	expected := "Also OK"455	if got != expected {456		t.Errorf("Expected to receive %s, got: %s", expected, got)457	}458	// We do not have another message pending so timer will459	// cancel the context.460	_, err = sub2.NextMsgWithContext(ctx)461	if err == nil {462		t.Fatal("Expected request with context to fail")463	}464	// Reported error is "context canceled" from Context package,465	// which is not a timeout error.466	type timeoutError interface {467		Timeout() bool468	}469	if _, ok := err.(timeoutError); ok {470		t.Errorf("Expected to not have a timeout error")471	}472	expected = `context canceled`473	if !strings.Contains(err.Error(), expected) {474		t.Errorf("Expected %q error, got: %q", expected, err.Error())475	}476}477func TestContextSubNextMsgWithDeadline(t *testing.T) {478	s := RunDefaultServer()479	defer s.Shutdown()480	nc := NewDefaultConnection(t)481	defer nc.Close()482	deadline := time.Now().Add(100 * time.Millisecond)483	ctx, cancelCB := context.WithDeadline(context.Background(), deadline)484	defer cancelCB() // should always be called, not discarded, to prevent context leak485	sub, err := nc.SubscribeSync("slow")486	if err != nil {487		t.Fatalf("Expected to be able to subscribe: %s", err)488	}489	for i := 0; i < 2; i++ {490		err := nc.Publish("slow", []byte("OK"))491		if err != nil {492			t.Fatalf("Expected publish to not fail: %s", err)493		}494		// Enough time to get a couple of messages495		time.Sleep(40 * time.Millisecond)496		msg, err := sub.NextMsgWithContext(ctx)497		if err != nil {498			t.Fatalf("Expected to receive message: %s", err)499		}500		got := string(msg.Data)501		expected := "OK"502		if got != expected {503			t.Errorf("Expected to receive %s, got: %s", expected, got)504		}505	}506	// Third message will fail because the context will be canceled by now507	_, err = sub.NextMsgWithContext(ctx)508	if err == nil {509		t.Fatal("Expected to fail receiving a message")510	}511	// Reported error is "context deadline exceeded" from Context package,512	// which implements net.Error Timeout interface.513	type timeoutError interface {514		Timeout() bool515	}516	timeoutErr, ok := err.(timeoutError)517	if !ok || !timeoutErr.Timeout() {518		t.Errorf("Expected to have a timeout error")519	}520	expected := `context deadline exceeded`521	if !strings.Contains(err.Error(), expected) {522		t.Errorf("Expected %q error, got: %q", expected, err.Error())523	}524}525func TestContextEncodedRequestWithTimeout(t *testing.T) {526	s := RunDefaultServer()527	defer s.Shutdown()528	nc := NewDefaultConnection(t)529	c, err := nats.NewEncodedConn(nc, nats.JSON_ENCODER)530	if err != nil {531		t.Fatalf("Unable to create encoded connection: %v", err)532	}533	defer c.Close()534	deadline := time.Now().Add(100 * time.Millisecond)535	ctx, cancelCB := context.WithDeadline(context.Background(), deadline)536	defer cancelCB() // should always be called, not discarded, to prevent context leak537	type request struct {538		Message string `json:"message"`539	}540	type response struct {541		Code int `json:"code"`542	}543	c.Subscribe("slow", func(_, reply string, req *request) {544		got := req.Message545		expected := "Hello"546		if got != expected {547			t.Errorf("Expected to receive request with %q, got %q", got, expected)548		}549		// simulates latency into the client so that timeout is hit.550		time.Sleep(40 * time.Millisecond)551		c.Publish(reply, &response{Code: 200})552	})553	for i := 0; i < 2; i++ {554		req := &request{Message: "Hello"}555		resp := &response{}556		err := c.RequestWithContext(ctx, "slow", req, resp)557		if err != nil {558			t.Fatalf("Expected encoded request with context to not fail: %s", err)559		}560		got := resp.Code561		expected := 200562		if got != expected {563			t.Errorf("Expected to receive %v, got: %v", expected, got)564		}565	}566	// A third request with latency would make the context567	// reach the deadline.568	req := &request{Message: "Hello"}569	resp := &response{}570	err = c.RequestWithContext(ctx, "slow", req, resp)571	if err == nil {572		t.Fatal("Expected request with context to reach deadline")573	}574	// Reported error is "context deadline exceeded" from Context package,575	// which implements net.Error Timeout interface.576	type timeoutError interface {577		Timeout() bool578	}579	timeoutErr, ok := err.(timeoutError)580	if !ok || !timeoutErr.Timeout() {581		t.Errorf("Expected to have a timeout error")582	}583	expected := `context deadline exceeded`584	if !strings.Contains(err.Error(), expected) {585		t.Errorf("Expected %q error, got: %q", expected, err.Error())586	}587}588func TestContextEncodedRequestWithTimeoutCanceled(t *testing.T) {589	s := RunDefaultServer()590	defer s.Shutdown()591	nc := NewDefaultConnection(t)592	c, err := nats.NewEncodedConn(nc, nats.JSON_ENCODER)593	if err != nil {594		t.Fatalf("Unable to create encoded connection: %v", err)595	}596	defer c.Close()597	ctx, cancelCB := context.WithTimeout(context.Background(), 100*time.Millisecond)598	defer cancelCB() // should always be called, not discarded, to prevent context leak599	type request struct {600		Message string `json:"message"`601	}602	type response struct {603		Code int `json:"code"`604	}605	c.Subscribe("fast", func(_, reply string, req *request) {606		got := req.Message607		expected := "Hello"608		if got != expected {609			t.Errorf("Expected to receive request with %q, got %q", got, expected)610		}611		// simulates latency into the client so that timeout is hit.612		time.Sleep(40 * time.Millisecond)613		c.Publish(reply, &response{Code: 200})614	})615	// Fast request should not fail616	req := &request{Message: "Hello"}617	resp := &response{}618	c.RequestWithContext(ctx, "fast", req, resp)619	expectedCode := 200620	if resp.Code != expectedCode {621		t.Errorf("Expected to receive %d, got: %d", expectedCode, resp.Code)622	}623	// Cancel the context already so that rest of requests fail.624	cancelCB()625	err = c.RequestWithContext(ctx, "fast", req, resp)626	if err == nil {627		t.Fatal("Expected request with timeout context to fail")628	}629	// Reported error is "context canceled" from Context package,630	// which is not a timeout error.631	type timeoutError interface {632		Timeout() bool633	}634	if _, ok := err.(timeoutError); ok {635		t.Errorf("Expected to not have a timeout error")636	}637	expected := `context canceled`638	if !strings.Contains(err.Error(), expected) {639		t.Errorf("Expected %q error, got: %q", expected, err.Error())640	}641	// 2nd request should fail again even if fast because context has already been canceled642	err = c.RequestWithContext(ctx, "fast", req, resp)643	if err == nil {644		t.Fatal("Expected request with timeout context to fail")645	}646}647func TestContextEncodedRequestWithCancel(t *testing.T) {648	s := RunDefaultServer()649	defer s.Shutdown()650	nc := NewDefaultConnection(t)651	c, err := nats.NewEncodedConn(nc, nats.JSON_ENCODER)652	if err != nil {653		t.Fatalf("Unable to create encoded connection: %v", err)654	}655	defer c.Close()656	ctx, cancelCB := context.WithCancel(context.Background())657	defer cancelCB() // should always be called, not discarded, to prevent context leak658	// timer which cancels the context though can also be arbitrarily extended659	expirationTimer := time.AfterFunc(100*time.Millisecond, func() {660		cancelCB()661	})662	type request struct {663		Message string `json:"message"`664	}665	type response struct {666		Code int `json:"code"`667	}668	c.Subscribe("slow", func(_, reply string, req *request) {669		got := req.Message670		expected := "Hello"671		if got != expected {672			t.Errorf("Expected to receive request with %q, got %q", got, expected)673		}674		// simulates latency into the client so that timeout is hit.675		time.Sleep(40 * time.Millisecond)676		c.Publish(reply, &response{Code: 200})677	})678	c.Subscribe("slower", func(_, reply string, req *request) {679		got := req.Message680		expected := "World"681		if got != expected {682			t.Errorf("Expected to receive request with %q, got %q", got, expected)683		}684		// we know this request will take longer so extend the timeout685		expirationTimer.Reset(100 * time.Millisecond)686		// slower reply which would have hit original timeout687		time.Sleep(90 * time.Millisecond)688		c.Publish(reply, &response{Code: 200})689	})690	for i := 0; i < 2; i++ {691		req := &request{Message: "Hello"}692		resp := &response{}693		err := c.RequestWithContext(ctx, "slow", req, resp)694		if err != nil {695			t.Fatalf("Expected encoded request with context to not fail: %s", err)696		}697		got := resp.Code698		expected := 200699		if got != expected {700			t.Errorf("Expected to receive %v, got: %v", expected, got)701		}702	}703	// A third request with latency would make the context704	// get canceled, but these reset the timer so deadline705	// gets extended:706	for i := 0; i < 10; i++ {707		req := &request{Message: "World"}708		resp := &response{}709		err := c.RequestWithContext(ctx, "slower", req, resp)710		if err != nil {711			t.Fatalf("Expected request with context to not fail: %s", err)712		}713		got := resp.Code714		expected := 200715		if got != expected {716			t.Errorf("Expected to receive %d, got: %d", expected, got)717		}718	}719	req := &request{Message: "Hello"}720	resp := &response{}721	// One more slow request will expire the timer and cause an error...722	err = c.RequestWithContext(ctx, "slow", req, resp)723	if err == nil {724		t.Fatal("Expected request with cancellation context to fail")725	}726	// ...though reported error is "context canceled" from Context package,727	// which is not a timeout error.728	type timeoutError interface {729		Timeout() bool730	}731	if _, ok := err.(timeoutError); ok {732		t.Errorf("Expected to not have a timeout error")733	}734	expected := `context canceled`735	if !strings.Contains(err.Error(), expected) {736		t.Errorf("Expected %q error, got: %q", expected, err.Error())737	}738}739func TestContextEncodedRequestWithDeadline(t *testing.T) {740	s := RunDefaultServer()741	defer s.Shutdown()742	nc := NewDefaultConnection(t)743	c, err := nats.NewEncodedConn(nc, nats.JSON_ENCODER)744	if err != nil {745		t.Fatalf("Unable to create encoded connection: %v", err)746	}747	defer c.Close()748	deadline := time.Now().Add(100 * time.Millisecond)749	ctx, cancelCB := context.WithDeadline(context.Background(), deadline)750	defer cancelCB() // should always be called, not discarded, to prevent context leak751	type request struct {752		Message string `json:"message"`753	}754	type response struct {755		Code int `json:"code"`756	}757	c.Subscribe("slow", func(_, reply string, req *request) {758		got := req.Message759		expected := "Hello"760		if got != expected {761			t.Errorf("Expected to receive request with %q, got %q", got, expected)762		}763		// simulates latency into the client so that timeout is hit.764		time.Sleep(40 * time.Millisecond)765		c.Publish(reply, &response{Code: 200})766	})767	for i := 0; i < 2; i++ {768		req := &request{Message: "Hello"}769		resp := &response{}770		err := c.RequestWithContext(ctx, "slow", req, resp)771		if err != nil {772			t.Fatalf("Expected encoded request with context to not fail: %s", err)773		}774		got := resp.Code775		expected := 200776		if got != expected {777			t.Errorf("Expected to receive %v, got: %v", expected, got)778		}779	}780	// A third request with latency would make the context781	// reach the deadline.782	req := &request{Message: "Hello"}783	resp := &response{}784	err = c.RequestWithContext(ctx, "slow", req, resp)785	if err == nil {786		t.Fatal("Expected request with context to reach deadline")787	}788	// Reported error is "context deadline exceeded" from Context package,789	// which implements net.Error Timeout interface.790	type timeoutError interface {791		Timeout() bool792	}793	timeoutErr, ok := err.(timeoutError)794	if !ok || !timeoutErr.Timeout() {795		t.Errorf("Expected to have a timeout error")796	}797	expected := `context deadline exceeded`798	if !strings.Contains(err.Error(), expected) {799		t.Errorf("Expected %q error, got: %q", expected, err.Error())800	}801}802func TestContextRequestConnClosed(t *testing.T) {803	s := RunDefaultServer()804	defer s.Shutdown()805	nc := NewDefaultConnection(t)806	ctx, cancelCB := context.WithCancel(context.Background())807	defer cancelCB()808	time.AfterFunc(100*time.Millisecond, func() {809		cancelCB()810	})811	nc.Close()812	_, err := nc.RequestWithContext(ctx, "foo", []byte(""))813	if err == nil {814		t.Fatal("Expected request to fail with error")815	}816	if err != nats.ErrConnectionClosed {817		t.Errorf("Expected request to fail with connection closed error: %s", err)818	}819}820func TestContextBadSubscription(t *testing.T) {821	s := RunDefaultServer()822	defer s.Shutdown()823	nc := NewDefaultConnection(t)824	defer nc.Close()825	ctx, cancelCB := context.WithCancel(context.Background())826	defer cancelCB()827	time.AfterFunc(100*time.Millisecond, func() {828		cancelCB()829	})830	sub, err := nc.Subscribe("foo", func(_ *nats.Msg) {})831	if err != nil {832		t.Fatalf("Expected to be able to subscribe: %s", err)833	}834	err = sub.Unsubscribe()835	if err != nil {836		t.Fatalf("Expected to be able to unsubscribe: %s", err)837	}838	_, err = sub.NextMsgWithContext(ctx)839	if err == nil {840		t.Fatal("Expected to fail getting next message with context")841	}842	if err != nats.ErrBadSubscription {843		t.Errorf("Expected request to fail with connection closed error: %s", err)844	}845}846func TestContextInvalid(t *testing.T) {847	s := RunDefaultServer()848	defer s.Shutdown()849	nc := NewDefaultConnection(t)850	c, err := nats.NewEncodedConn(nc, nats.JSON_ENCODER)851	if err != nil {852		t.Fatalf("Unable to create encoded connection: %v", err)853	}854	defer c.Close()855	//lint:ignore SA1012 testing that passing nil fails856	_, err = nc.RequestWithContext(nil, "foo", []byte(""))857	if err == nil {858		t.Fatal("Expected request to fail with error")859	}860	if err != nats.ErrInvalidContext {861		t.Errorf("Expected request to fail with connection closed error: %s", err)862	}863	sub, err := nc.Subscribe("foo", func(_ *nats.Msg) {})864	if err != nil {865		t.Fatalf("Expected to be able to subscribe: %s", err)866	}867	//lint:ignore SA1012 testing that passing nil fails868	_, err = sub.NextMsgWithContext(nil)869	if err == nil {870		t.Fatal("Expected request to fail with error")871	}872	if err != nats.ErrInvalidContext {873		t.Errorf("Expected request to fail with connection closed error: %s", err)874	}875	type request struct {876		Message string `json:"message"`877	}878	type response struct {879		Code int `json:"code"`880	}881	req := &request{Message: "Hello"}882	resp := &response{}883	//lint:ignore SA1012 testing that passing nil fails884	err = c.RequestWithContext(nil, "slow", req, resp)885	if err == nil {886		t.Fatal("Expected request to fail with error")887	}888	if err != nats.ErrInvalidContext {889		t.Errorf("Expected request to fail with invalid context: %s", err)890	}891}892func TestFlushWithContext(t *testing.T) {893	s := RunDefaultServer()894	defer s.Shutdown()895	nc := NewDefaultConnection(t)896	defer nc.Close()897	ctx := context.Background()898	// No context should error.899	//lint:ignore SA1012 testing that passing nil fails900	if err := nc.FlushWithContext(nil); err != nats.ErrInvalidContext {901		t.Fatalf("Expected '%v', got '%v'", nats.ErrInvalidContext, err)902	}903	// A context with no deadline set should error also.904	if err := nc.FlushWithContext(ctx); err != nats.ErrNoDeadlineContext {905		t.Fatalf("Expected '%v', got '%v'", nats.ErrNoDeadlineContext, err)906	}907	dctx, cancel := context.WithTimeout(ctx, 0)908	defer cancel()909	// A context with a deadline should return when expired.910	if err := nc.FlushWithContext(dctx); err != context.DeadlineExceeded {911		t.Fatalf("Expected '%v', got '%v'", context.DeadlineExceeded, err)912	}913}914func TestUnsubscribeAndNextMsgWithContext(t *testing.T) {915	s := RunDefaultServer()916	defer s.Shutdown()917	nc := NewDefaultConnection(t)918	defer nc.Close()919	ctx, cancelCB := context.WithCancel(context.Background())920	defer cancelCB() // should always be called, not discarded, to prevent context leak921	sub, err := nc.SubscribeSync("foo")...

...22	"time"23	"google.golang.org/grpc/internal/grpctest"24)25const (26	testCacheTimeout = 100 * time.Millisecond27)28type s struct {29	grpctest.Tester30}31func Test(t *testing.T) {32	grpctest.RunSubTests(t, s{})33}34func (c *TimeoutCache) getForTesting(key interface{}) (*cacheEntry, bool) {35	c.mu.Lock()36	defer c.mu.Unlock()37	r, ok := c.cache[key]38	return r, ok39}40// TestCacheExpire attempts to add an entry to the cache and verifies that it41// was added successfully. It then makes sure that on timeout, it's removed and42// the associated callback is called.43func (s) TestCacheExpire(t *testing.T) {44	const k, v = 1, "1"45	c := NewTimeoutCache(testCacheTimeout)46	callbackChan := make(chan struct{})47	c.Add(k, v, func() { close(callbackChan) })48	if gotV, ok := c.getForTesting(k); !ok || gotV.item != v {49		t.Fatalf("After Add(), before timeout, from cache got: %v, %v, want %v, %v", gotV.item, ok, v, true)50	}51	select {52	case <-callbackChan:53	case <-time.After(testCacheTimeout * 2):54		t.Fatalf("timeout waiting for callback")55	}56	if _, ok := c.getForTesting(k); ok {57		t.Fatalf("After Add(), after timeout, from cache got: _, %v, want _, %v", ok, false)58	}59}60// TestCacheRemove attempts to remove an existing entry from the cache and61// verifies that the entry is removed and the associated callback is not62// invoked.63func (s) TestCacheRemove(t *testing.T) {64	const k, v = 1, "1"65	c := NewTimeoutCache(testCacheTimeout)66	callbackChan := make(chan struct{})67	c.Add(k, v, func() { close(callbackChan) })68	if got, ok := c.getForTesting(k); !ok || got.item != v {69		t.Fatalf("After Add(), before timeout, from cache got: %v, %v, want %v, %v", got.item, ok, v, true)70	}71	time.Sleep(testCacheTimeout / 2)72	gotV, gotOK := c.Remove(k)73	if !gotOK || gotV != v {74		t.Fatalf("After Add(), before timeout, Remove() got: %v, %v, want %v, %v", gotV, gotOK, v, true)75	}76	if _, ok := c.getForTesting(k); ok {77		t.Fatalf("After Add(), before timeout, after Remove(), from cache got: _, %v, want _, %v", ok, false)78	}79	select {80	case <-callbackChan:81		t.Fatalf("unexpected callback after retrieve")82	case <-time.After(testCacheTimeout * 2):83	}84}85// TestCacheClearWithoutCallback attempts to clear all entries from the cache86// and verifies that the associated callbacks are not invoked.87func (s) TestCacheClearWithoutCallback(t *testing.T) {88	var values []string89	const itemCount = 390	for i := 0; i < itemCount; i++ {91		values = append(values, strconv.Itoa(i))92	}93	c := NewTimeoutCache(testCacheTimeout)94	done := make(chan struct{})95	defer close(done)96	callbackChan := make(chan struct{}, itemCount)97	for i, v := range values {98		callbackChanTemp := make(chan struct{})99		c.Add(i, v, func() { close(callbackChanTemp) })100		go func() {101			select {102			case <-callbackChanTemp:103				callbackChan <- struct{}{}104			case <-done:105			}106		}()107	}108	for i, v := range values {109		if got, ok := c.getForTesting(i); !ok || got.item != v {110			t.Fatalf("After Add(), before timeout, from cache got: %v, %v, want %v, %v", got.item, ok, v, true)111		}112	}113	time.Sleep(testCacheTimeout / 2)114	c.Clear(false)115	for i := range values {116		if _, ok := c.getForTesting(i); ok {117			t.Fatalf("After Add(), before timeout, after Remove(), from cache got: _, %v, want _, %v", ok, false)118		}119	}120	select {121	case <-callbackChan:122		t.Fatalf("unexpected callback after Clear")123	case <-time.After(testCacheTimeout * 2):124	}125}126// TestCacheClearWithCallback attempts to clear all entries from the cache and127// verifies that the associated callbacks are invoked.128func (s) TestCacheClearWithCallback(t *testing.T) {129	var values []string130	const itemCount = 3131	for i := 0; i < itemCount; i++ {132		values = append(values, strconv.Itoa(i))133	}134	c := NewTimeoutCache(time.Hour)135	testDone := make(chan struct{})136	defer close(testDone)137	var wg sync.WaitGroup138	wg.Add(itemCount)139	for i, v := range values {140		callbackChanTemp := make(chan struct{})141		c.Add(i, v, func() { close(callbackChanTemp) })142		go func() {143			defer wg.Done()144			select {145			case <-callbackChanTemp:146			case <-testDone:147			}148		}()149	}150	allGoroutineDone := make(chan struct{}, itemCount)151	go func() {152		wg.Wait()153		close(allGoroutineDone)154	}()155	for i, v := range values {156		if got, ok := c.getForTesting(i); !ok || got.item != v {157			t.Fatalf("After Add(), before timeout, from cache got: %v, %v, want %v, %v", got.item, ok, v, true)158		}159	}160	time.Sleep(testCacheTimeout / 2)161	c.Clear(true)162	for i := range values {163		if _, ok := c.getForTesting(i); ok {164			t.Fatalf("After Add(), before timeout, after Remove(), from cache got: _, %v, want _, %v", ok, false)165		}166	}167	select {168	case <-allGoroutineDone:169	case <-time.After(testCacheTimeout * 2):170		t.Fatalf("timeout waiting for all callbacks")171	}172}173// TestCacheRetrieveTimeoutRace simulates the case where an entry's timer fires174// around the same time that Remove() is called for it. It verifies that there175// is no deadlock.176func (s) TestCacheRetrieveTimeoutRace(t *testing.T) {177	c := NewTimeoutCache(time.Nanosecond)178	done := make(chan struct{})179	go func() {180		for i := 0; i < 1000; i++ {181			// Add starts a timer with 1 ns timeout, then remove will race182			// with the timer.183			c.Add(i, strconv.Itoa(i), func() {})184			c.Remove(i)185		}186		close(done)187	}()188	select {189	case <-time.After(time.Second):190		t.Fatalf("Test didn't finish within 1 second. Deadlock")191	case <-done:...

1import (2func main() {3	c1 := make(chan string, 1)4	go func() {5		time.Sleep(time.Second * 2)6	}()7	select {8		fmt.Println(res)9	case <-time.After(time.Second * 1):10		fmt.Println("timeout 1")11	}12	c2 := make(chan string, 1)13	go func() {14		time.Sleep(time.Second * 2)15	}()16	select {17		fmt.Println(res)18	case <-time.After(time.Second * 3):19		fmt.Println("timeout 2")20	}21}22import (23func main() {24	c1 := make(chan string, 1)25	go func() {26		for {27			time.Sleep(time.Second * 2)28		}29	}()30	for {31		select {32			fmt.Println(res)33		case <-time.After(time.Second * 1):34			fmt.Println("timeout 1")

1import (2func main() {3	c := make(chan string, 1)4	go func() {5		time.Sleep(2 * time.Second)6	}()7	select {8		fmt.Println(res)9	case <-time.After(1 * time.Second):10		fmt.Println("timeout 1")11	}12}

1import (2func main() {3	c := make(chan int)4	go func() {5		time.Sleep(2 * time.Second)6	}()7	select {8		fmt.Println("received", m)9	case <-time.After(1 * time.Second):10		fmt.Println("timeout")11	}12}13import (14func main() {15	c := make(chan int)16	go func() {17		time.Sleep(1 * time.Second)18	}()19	select {20		fmt.Println("received", m)21	case <-time.After(2 * time.Second):22		fmt.Println("timeout")23	}24}25import (26func main() {27	c := make(chan int)28	go func() {29		time.Sleep(2 * time.Second)30	}()31	select {32		fmt.Println("received", m)33	case <-time.After(3 * time.Second):34		fmt.Println("timeout")35	}36}37import (38func main() {39	c := make(chan int)40	go func() {41		time.Sleep(3 * time.Second)42	}()43	select {44		fmt.Println("received", m)45	case <-time.After(2 * time.Second):46		fmt.Println("timeout")47	}48}49import (50func main() {51	c := make(chan int)52	go func() {53		time.Sleep(3 * time.Second)54	}()55	select {56		fmt.Println("received", m)57	case <-time.After(4 * time.Second):58		fmt.Println("timeout")59	}60}

1func main() {2    timeout := time.After(1 * time.Second)3    done := make(chan bool)4    go func() {5        time.Sleep(2 * time.Second)6    }()7    select {8        fmt.Println("Work complete")9        fmt.Println("Timed out")10    }11}12func main() {13    timeout := time.After(1 * time.Second)14    done := make(chan bool)15    go func() {16        time.Sleep(500 * time.Millisecond)17    }()18    select {19        fmt.Println("Work complete")20        fmt.Println("Timed out")21        fmt.Println("Default case")22    }23}24func main() {25    timeout := time.After(1 * time.Second)

1import (2func main() {3	fmt.Println("Start")4	timeout := time.After(2 * time.Second)5	fmt.Println("End")6}7import (8func main() {9	fmt.Println("Start")10	timer := time.NewTimer(2 * time.Second)11	fmt.Println("End")12}13import (14func main() {15	fmt.Println("Start")16	time.Sleep(2 * time.Second)17	fmt.Println("End")18}19import (20func main() {21	fmt.Println("Start")22	time.AfterFunc(2*time.Second, func() {23		fmt.Println("End")24	})25	time.Sleep(3 * time.Second)26}27import (28func main() {29	fmt.Println("Start")30	ticker := time.NewTicker(2 * time.Second)