改名
This commit is contained in:
+6
-1
@@ -18,6 +18,7 @@ use RdKafka\Conf;
|
|||||||
use RdKafka\TopicConf;
|
use RdKafka\TopicConf;
|
||||||
use ReflectionException;
|
use ReflectionException;
|
||||||
use Snowflake\Abstracts\Component;
|
use Snowflake\Abstracts\Component;
|
||||||
|
use Snowflake\Event;
|
||||||
use Snowflake\Exception\NotFindClassException;
|
use Snowflake\Exception\NotFindClassException;
|
||||||
use Snowflake\Snowflake;
|
use Snowflake\Snowflake;
|
||||||
|
|
||||||
@@ -105,6 +106,10 @@ class Producer extends Component
|
|||||||
$topic = $rk->newTopic($this->_topic, $this->topicConf);
|
$topic = $rk->newTopic($this->_topic, $this->topicConf);
|
||||||
$topic->produce(RD_KAFKA_PARTITION_UA, 0, $message, $key);
|
$topic->produce(RD_KAFKA_PARTITION_UA, 0, $message, $key);
|
||||||
$rk->poll($timeout);
|
$rk->poll($timeout);
|
||||||
$rk->flush($timeout);
|
|
||||||
|
$event = Snowflake::app()->getEvent();
|
||||||
|
$event->on(Event::EVENT_AFTER_REQUEST, function () use ($rk, $timeout) {
|
||||||
|
$rk->flush($timeout);
|
||||||
|
});
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user