@@ -58,59 +58,68 @@ func (p *Processor) Start() error {
5858}
5959
6060func (p * Processor ) processOneByOne () error {
61- ctx , cancel := context .WithTimeout (context .Background (), p .processTickTimeout )
62- defer cancel ()
61+ f := func (rp connector.RequestPerformer ) error {
62+ ctx , cancel := context .WithTimeout (context .Background (), p .processTickTimeout )
63+ defer cancel ()
6364
64- coreClient := core .NewClient (p .qubicConnector )
65- tickInfo , err := coreClient .GetTickInfo (ctx )
66- if err != nil {
67- return errors .Wrap (err , "getting tick info" )
68- }
69-
70- lastTick , err := p .getLastProcessedTick (ctx , tickInfo )
71- if err != nil {
72- return errors .Wrap (err , "getting last processed tick" )
73- }
65+ coreClient := core .NewClient (rp )
66+ tickInfo , err := coreClient .GetTickInfo (ctx )
67+ if err != nil {
68+ return errors .Wrap (err , "getting tick info" )
69+ }
7470
75- nextTick , err := p .getNextProcessingTick (ctx , lastTick , tickInfo )
76- if err != nil {
77- return errors .Wrap (err , "getting next processing tick" )
78- }
79- log .Printf ("Next tick to process: %d\n " , nextTick .TickNumber )
71+ lastTick , err := p .getLastProcessedTick (ctx , tickInfo )
72+ if err != nil {
73+ return errors .Wrap (err , "getting last processed tick" )
74+ }
8075
81- if tickInfo .Tick < nextTick .TickNumber {
82- err = newTickInTheFutureError (nextTick .TickNumber , tickInfo .Tick )
83- return err
84- }
76+ nextTick , err := p .getNextProcessingTick (ctx , lastTick , tickInfo )
77+ if err != nil {
78+ return errors .Wrap (err , "getting next processing tick" )
79+ }
80+ log .Printf ("Next tick to process: %d\n " , nextTick .TickNumber )
8581
86- eventsClient := events .NewClient (p .qubicConnector , p .passcodes )
87- start := time .Now ()
88- tickEvents , err := eventsClient .GetTickEvents (context .Background (), nextTick .TickNumber )
89- if err != nil {
90- return errors .Wrap (err , "getting tick events" )
91- }
92- end := time .Now ()
82+ if tickInfo .Tick < nextTick .TickNumber {
83+ err = newTickInTheFutureError (nextTick .TickNumber , tickInfo .Tick )
84+ return err
85+ }
9386
94- err = p .eventsStore .SetTickEvents (nextTick .TickNumber , tickEvents )
95- if err != nil {
96- return errors .Wrap (err , "setting tick events" )
97- }
87+ eventsClient := events .NewClient (rp , p .passcodes )
88+ start := time .Now ()
89+ tickEvents , err := eventsClient .GetTickEvents (context .Background (), nextTick .TickNumber )
90+ if err != nil {
91+ return errors .Wrap (err , "getting tick events" )
92+ }
93+ end := time .Now ()
9894
99- err = p .eventsStore .SetTickProcessTime (nextTick .TickNumber , uint64 ( end . Sub ( start ). Seconds ()) )
100- if err != nil {
101- return errors .Wrap (err , "setting tick process time " )
102- }
95+ err = p .eventsStore .SetTickEvents (nextTick .TickNumber , tickEvents )
96+ if err != nil {
97+ return errors .Wrap (err , "setting tick events " )
98+ }
10399
104- err = p .processStatus ( ctx , lastTick , nextTick )
105- if err != nil {
106- return errors .Wrapf (err , "processing status for lastTick %+v and nextTick %+v" , lastTick , nextTick )
107- }
100+ err = p .eventsStore . SetTickProcessTime ( nextTick . TickNumber , uint64 ( end . Sub ( start ). Seconds ()) )
101+ if err != nil {
102+ return errors .Wrap (err , "setting tick process time" )
103+ }
108104
109- if p .isPubSubEnabled {
110- err = p .redisPubSubClient .PublishTickEvents (ctx , tickEvents )
105+ err = p .processStatus (ctx , lastTick , nextTick )
111106 if err != nil {
112- return errors .Wrap (err , "publishing tick events" )
107+ return errors .Wrapf (err , "processing status for lastTick %+v and nextTick %+v" , lastTick , nextTick )
113108 }
109+
110+ if p .isPubSubEnabled {
111+ err = p .redisPubSubClient .PublishTickEvents (ctx , tickEvents )
112+ if err != nil {
113+ return errors .Wrap (err , "publishing tick events" )
114+ }
115+ }
116+
117+ return nil
118+ }
119+
120+ err := p .qubicConnector .WithConnection (f )
121+ if err != nil {
122+ return errors .Wrap (err , "performing WithConnection logic" )
114123 }
115124
116125 return nil
@@ -142,7 +151,7 @@ func (p *Processor) getNextProcessingTick(ctx context.Context, lastTick *eventsp
142151}
143152
144153func (p * Processor ) getLastProcessedTick (ctx context.Context , currentTickInfo * qubicpb.TickInfo ) (* eventspb.ProcessedTick , error ) {
145- lastTick , err := p .eventsStore .GetLastProcessedTick ( ctx )
154+ lastTick , err := p .eventsStore .FetchLastProcessedTick ( )
146155 if err != nil {
147156 //handles first run of the events processor where there is nothing in storage
148157 // in this case last tick is 0 and epoch is current tick info epoch
0 commit comments