Update PublishToSubject to return ErrAckTimeout

client.go

@@ -990,6 +990,16 @@ func (c *client) PublishToSubject(ctx context.Context, subject string, value []b
 		AckPolicy:     opts.AckPolicy.toProto(),
 	}
 
+	// Setup ack timeout.
+	var (
+		cancel func()
+		_, ok  = ctx.Deadline()
+	)
+	if !ok {
+		ctx, cancel = context.WithTimeout(ctx, c.opts.AckWaitTime)
+		defer cancel()
+	}
+
 	var ack *proto.Ack
 	err := c.doResilientRPC(func(client proto.APIClient) error {
 		resp, err := client.PublishToSubject(ctx, req)
@@ -998,6 +1008,9 @@ func (c *client) PublishToSubject(ctx context.Context, subject string, value []b
 		}
 		return err
 	})
+	if status.Code(err) == codes.DeadlineExceeded {
+		err = ErrAckTimeout
+	}
 	return ackFromProto(ack), err
 }
 

client_test.go

@@ -1102,6 +1102,23 @@ func TestPublishToSubject(t *testing.T) {
 	require.Equal(t, proto.AckPolicy_LEADER, req.AckPolicy)
 }
 
+func TestPublishToSubjectAckTimeout(t *testing.T) {
+	server := newMockServer()
+	defer server.Stop(t)
+	port := server.Start(t)
+
+	server.SetupMockResponse(new(proto.FetchMetadataResponse))
+
+	client, err := Connect([]string{fmt.Sprintf("localhost:%d", port)})
+	require.NoError(t, err)
+	defer client.Close()
+
+	server.SetupMockPublishToSubjectError(status.Error(codes.DeadlineExceeded, "deadline exceeded"))
+
+	_, err = client.PublishToSubject(context.Background(), "foo", []byte("hello"))
+	require.Equal(t, ErrAckTimeout, err)
+}
+
 func TestFetchMetadata(t *testing.T) {
 	server := newMockServer()
 	defer server.Stop(t)