makhmutov
commented
2 years ago пробывал вариант с cancelFunc, так же получаю ResourceExhausted на второй итерации
package tinkoff
import (
"context"
"crypto/tls"
"fmt"
"github.com/rs/zerolog/log"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/metadata"
"xxx/config"
pb "xxx/investapi"
"io"
"sync"
"testing"
"time"
)
const ReconnectCount = 2
const IdleTIme = time.Second * 10
const SleepTime = time.Second * 5
func TestGracefullShutdownWithOneConnection(t *testing.T) {
tinkoffToken := config.GetConfigFromEnv().TinkoffToken
md := metadata.New(map[string]string{"Authorization": "Bearer " + tinkoffToken})
for ci := 0; ci < ReconnectCount; ci++ {
func() {
fmt.Println("Create connection", ci)
conn, err := grpc.Dial("invest-public-api.tinkoff.ru:443", grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{InsecureSkipVerify: true})), grpc.WithBlock()) //FIXME
if err != nil {
log.Fatal().Msgf("err on grpc.Dial: %v", err)
}
defer func() {
fmt.Println("Close connection...", ci)
conn.Close()
}()
var wg sync.WaitGroup
tariff, err := pb.NewUsersServiceClient(conn).GetUserTariff(metadata.NewOutgoingContext(context.TODO(), md), &pb.GetUserTariffRequest{})
if err != nil {
log.Fatal().Msgf("err on client.GetUserTariff: %v", err)
}
var streamLimit int32
for _, limit := range tariff.StreamLimits {
for _, stream := range limit.Streams {
if stream == "tinkoff.public.invest.api.contract.v1.MarketDataStreamService/MarketDataStream" {
streamLimit = limit.Limit
}
}
}
if streamLimit == 0 {
log.Fatal().Msgf("streamLimit == 0")
}
fmt.Printf("streamLimit = %d\n", streamLimit)
for si := int32(0); si < streamLimit; si++ {
wg.Add(1)
streamIndex := si
go func() {
defer wg.Done()
fmt.Printf("Create stream[%d] of connection[%d] \n", streamIndex, ci)
ctx, cancelFunc := context.WithCancel(context.Background())
streamClient, err := pb.NewMarketDataStreamServiceClient(conn).MarketDataStream(metadata.NewOutgoingContext(ctx, md))
if err != nil {
log.Fatal().Msgf("err on client.MarketDataStream: %v", err)
}
err = streamClient.Send(&pb.MarketDataRequest{
Payload: &pb.MarketDataRequest_SubscribeTradesRequest{
SubscribeTradesRequest: &pb.SubscribeTradesRequest{
SubscriptionAction: pb.SubscriptionAction_SUBSCRIPTION_ACTION_SUBSCRIBE,
Instruments: []*pb.TradeInstrument{
{
Figi: "BBG002B04MT8",
},
},
},
},
})
go func() {
for {
in, err := streamClient.Recv()
if err == io.EOF {
return
}
if err != nil {
log.Err(err).Msgf("err on streamClient.Recv: %v", err)
return
}
fmt.Println(in)
}
}()
if err != nil {
log.Fatal().Msgf("err on streamClient.Send: %v", err)
}
time.Sleep(IdleTIme)
cancelFunc()
err = streamClient.CloseSend()
if err != nil {
log.Fatal().Msgf("err on streamClient.CloseSend: %v", err)
}
}()
}
wg.Wait()
time.Sleep(SleepTime)
fmt.Println()
}()
}
}
betslus1
malyginvv
nonamegithub
AlexanderVolkovTCS
Вывод