(ctx context.Context, seqId int32, iprot, oprot thrift.TProtocol)
| 24027 | } |
| 24028 | |
| 24029 | func (p *iClientRPCServiceProcessorInsertRecords) Process(ctx context.Context, seqId int32, iprot, oprot thrift.TProtocol) (success bool, err thrift.TException) { |
| 24030 | args := IClientRPCServiceInsertRecordsArgs{} |
| 24031 | var err2 error |
| 24032 | if err2 = args.Read(ctx, iprot); err2 != nil { |
| 24033 | iprot.ReadMessageEnd(ctx) |
| 24034 | x := thrift.NewTApplicationException(thrift.PROTOCOL_ERROR, err2.Error()) |
| 24035 | oprot.WriteMessageBegin(ctx, "insertRecords", thrift.EXCEPTION, seqId) |
| 24036 | x.Write(ctx, oprot) |
| 24037 | oprot.WriteMessageEnd(ctx) |
| 24038 | oprot.Flush(ctx) |
| 24039 | return false, thrift.WrapTException(err2) |
| 24040 | } |
| 24041 | iprot.ReadMessageEnd(ctx) |
| 24042 | |
| 24043 | tickerCancel := func() {} |
| 24044 | // Start a goroutine to do server side connectivity check. |
| 24045 | if thrift.ServerConnectivityCheckInterval > 0 { |
| 24046 | var cancel context.CancelFunc |
| 24047 | ctx, cancel = context.WithCancel(ctx) |
| 24048 | defer cancel() |
| 24049 | var tickerCtx context.Context |
| 24050 | tickerCtx, tickerCancel = context.WithCancel(context.Background()) |
| 24051 | defer tickerCancel() |
| 24052 | go func(ctx context.Context, cancel context.CancelFunc) { |
| 24053 | ticker := time.NewTicker(thrift.ServerConnectivityCheckInterval) |
| 24054 | defer ticker.Stop() |
| 24055 | for { |
| 24056 | select { |
| 24057 | case <-ctx.Done(): |
| 24058 | return |
| 24059 | case <-ticker.C: |
| 24060 | if !iprot.Transport().IsOpen() { |
| 24061 | cancel() |
| 24062 | return |
| 24063 | } |
| 24064 | } |
| 24065 | } |
| 24066 | }(tickerCtx, cancel) |
| 24067 | } |
| 24068 | |
| 24069 | result := IClientRPCServiceInsertRecordsResult{} |
| 24070 | var retval *common.TSStatus |
| 24071 | if retval, err2 = p.handler.InsertRecords(ctx, args.Req); err2 != nil { |
| 24072 | tickerCancel() |
| 24073 | if err2 == thrift.ErrAbandonRequest { |
| 24074 | return false, thrift.WrapTException(err2) |
| 24075 | } |
| 24076 | x := thrift.NewTApplicationException(thrift.INTERNAL_ERROR, "Internal error processing insertRecords: " + err2.Error()) |
| 24077 | oprot.WriteMessageBegin(ctx, "insertRecords", thrift.EXCEPTION, seqId) |
| 24078 | x.Write(ctx, oprot) |
| 24079 | oprot.WriteMessageEnd(ctx) |
| 24080 | oprot.Flush(ctx) |
| 24081 | return true, thrift.WrapTException(err2) |
| 24082 | } else { |
| 24083 | result.Success = retval |
| 24084 | } |
| 24085 | tickerCancel() |
| 24086 | if err2 = oprot.WriteMessageBegin(ctx, "insertRecords", thrift.REPLY, seqId); err2 != nil { |
nothing calls this directly
no test coverage detected