File tree 1 file changed +8
-11
lines changed
1 file changed +8
-11
lines changed Original file line number Diff line number Diff line change @@ -99,18 +99,17 @@ public function receive(int $timeout = 0): ?Message
99
99
$ this ->subscribed = true ;
100
100
}
101
101
102
- $ message = null ;
103
102
if ($ timeout > 0 ) {
104
- $ message = $ this ->doReceive ($ timeout );
105
- } else {
106
- while ( true ) {
107
- if ( $ message = $ this -> doReceive ( 500 ) ) {
108
- break ;
109
- }
103
+ return $ this ->doReceive ($ timeout );
104
+ }
105
+
106
+ while ( true ) {
107
+ if ( $ message = $ this -> doReceive ( 500 )) {
108
+ return $ message ;
110
109
}
111
110
}
112
111
113
- return $ message ;
112
+ return null ;
114
113
}
115
114
116
115
/**
@@ -162,7 +161,7 @@ private function doReceive(int $timeout): ?RdKafkaMessage
162
161
switch ($ kafkaMessage ->err ) {
163
162
case RD_KAFKA_RESP_ERR__PARTITION_EOF :
164
163
case RD_KAFKA_RESP_ERR__TIMED_OUT :
165
- break ;
164
+ return null ;
166
165
case RD_KAFKA_RESP_ERR_NO_ERROR :
167
166
$ message = $ this ->serializer ->toMessage ($ kafkaMessage ->payload );
168
167
$ message ->setKey ($ kafkaMessage ->key );
@@ -180,7 +179,5 @@ private function doReceive(int $timeout): ?RdKafkaMessage
180
179
throw new \LogicException ($ kafkaMessage ->errstr (), $ kafkaMessage ->err );
181
180
break ;
182
181
}
183
-
184
- return null ;
185
182
}
186
183
}
You can’t perform that action at this time.
0 commit comments