<?php
 namespace MongoDB\Tests\SpecTests;
 use IteratorIterator;
 use MongoDB\Client;
 use MongoDB\Collection;
 use MongoDB\Driver\Command;
 use MongoDB\Driver\Exception\BulkWriteException;
 use MongoDB\Driver\Exception\Exception as DriverException;
 use MongoDB\Driver\ReadPreference;
 use MongoDB\Driver\Server;
 use MongoDB\Driver\WriteConcern;
 use MongoDB\Operation\BulkWrite;
 use MongoDB\Tests\CommandObserver;
 use Symfony\Bridge\PhpUnit\SetUpTearDownTrait;
 use UnexpectedValueException;
 use function current;
 class PrimaryStepDownSpecTest extends FunctionalTestCase {
 use SetUpTearDownTrait;
 const INTERRUPTED_AT_SHUTDOWN = 11600;
 const NOT_MASTER = 10107;
 const SHUTDOWN_IN_PROGRESS = 91;
 private $client;
 private $collection;
 private function doSetUp() {
 parent::setUp();
 $this->client = new Client(static::getUri(), ['retryWrites' => false, 'heartbeatFrequencyMS' => 500, 'serverSelectionTimeoutMS' => 20000, 'serverSelectionTryOnce' => false]);
 $this->dropAndRecreateCollection();
 $this->collection = $this->client->selectCollection($this->getDatabaseName(), $this->getCollectionName());
 }
 public function testNotMasterKeepsConnectionPool() {
 $runOn = [(object) ['minServerVersion' => '4.1.11', 'topology' => [self::TOPOLOGY_REPLICASET]]];
 $this->checkServerRequirements($runOn);
 $this->configureFailPoint([ 'configureFailPoint' => 'failCommand', 'mode' => ['times' => 1], 'data' => [ 'failCommands' => ['insert'], 'errorCode' => self::NOT_MASTER, ], ]);
 $totalConnectionsCreated = $this->getTotalConnectionsCreated();
 try {
 $this->insertDocuments(1);
 }
 catch (BulkWriteException $e) {
 $this->assertSame(self::NOT_MASTER, $e->getCode());
 }
 $result = $this->insertDocuments(1);
 $this->assertSame(1, $result->getInsertedCount());
 $this->assertSame($totalConnectionsCreated, $this->getTotalConnectionsCreated());
 }
 public function testNotMasterResetConnectionPool() {
 $runOn = [(object) ['minServerVersion' => '4.0.0', 'maxServerVersion' => '4.0.999', 'topology' => [self::TOPOLOGY_REPLICASET]]];
 $this->checkServerRequirements($runOn);
 $this->configureFailPoint([ 'configureFailPoint' => 'failCommand', 'mode' => ['times' => 1], 'data' => [ 'failCommands' => ['insert'], 'errorCode' => self::NOT_MASTER, ], ]);
 $totalConnectionsCreated = $this->getTotalConnectionsCreated();
 try {
 $this->insertDocuments(1);
 }
 catch (BulkWriteException $e) {
 $this->assertSame(self::NOT_MASTER, $e->getCode());
 }
 $this->assertSame($totalConnectionsCreated + 1, $this->getTotalConnectionsCreated());
 $result = $this->insertDocuments(1);
 $this->assertSame(1, $result->getInsertedCount());
 }
 public function testShutdownResetConnectionPool() {
 $runOn = [(object) ['minServerVersion' => '4.0.0']];
 $this->checkServerRequirements($runOn);
 $this->configureFailPoint([ 'configureFailPoint' => 'failCommand', 'mode' => ['times' => 1], 'data' => [ 'failCommands' => ['insert'], 'errorCode' => self::SHUTDOWN_IN_PROGRESS, ], ]);
 $totalConnectionsCreated = $this->getTotalConnectionsCreated();
 try {
 $this->insertDocuments(1);
 }
 catch (BulkWriteException $e) {
 $this->assertSame(self::SHUTDOWN_IN_PROGRESS, $e->getCode());
 }
 $this->assertSame($totalConnectionsCreated + 1, $this->getTotalConnectionsCreated());
 $result = $this->insertDocuments(1);
 $this->assertSame(1, $result->getInsertedCount());
 }
 public function testInterruptedAtShutdownResetConnectionPool() {
 $runOn = [(object) ['minServerVersion' => '4.0.0']];
 $this->checkServerRequirements($runOn);
 $this->configureFailPoint([ 'configureFailPoint' => 'failCommand', 'mode' => ['times' => 1], 'data' => [ 'failCommands' => ['insert'], 'errorCode' => self::INTERRUPTED_AT_SHUTDOWN, ], ]);
 $totalConnectionsCreated = $this->getTotalConnectionsCreated();
 try {
 $this->insertDocuments(1);
 }
 catch (BulkWriteException $e) {
 $this->assertSame(self::INTERRUPTED_AT_SHUTDOWN, $e->getCode());
 }
 $this->assertSame($totalConnectionsCreated + 1, $this->getTotalConnectionsCreated());
 $result = $this->insertDocuments(1);
 $this->assertSame(1, $result->getInsertedCount());
 }
 public function testGetMoreIteration() {
 $this->markTestSkipped('Test causes subsequent failures in other tests (see PHPLIB-471)');
 $runOn = [(object) ['minServerVersion' => '4.1.11', 'topology' => [self::TOPOLOGY_REPLICASET]]];
 $this->checkServerRequirements($runOn);
 $this->insertDocuments(5);
 $cursor = $this->collection->find([], ['batchSize' => 2]);
 $iterator = new IteratorIterator($cursor);
 $iterator->rewind();
 $this->assertTrue($iterator->valid());
 $iterator->next();
 $this->assertTrue($iterator->valid());
 $totalConnectionsCreated = $this->getTotalConnectionsCreated();
 $primary = $this->client->getManager()->selectServer(new ReadPreference(ReadPreference::RP_PRIMARY));
 $primary->executeCommand('admin', new Command(['replSetStepDown' => 5, 'force' => true]));
 $events = [];
 $observer = new CommandObserver();
 $observer->observe( function () use ($iterator) {
 $iterator->next();
 }
, function ($event) use (&$events) {
 $events[] = $event;
 }
 );
 $this->assertTrue($iterator->valid());
 $this->assertCount(1, $events);
 $this->assertSame('getMore', $events[0]['started']->getCommandName());
 $this->assertSame($totalConnectionsCreated, $this->getTotalConnectionsCreated($cursor->getServer()));
 $this->waitForMasterReelection();
 }
 private function insertDocuments($count) {
 $operations = [];
 for ($i = 1;
 $i <= $count;
 $i++) {
 $operations[] = [ BulkWrite::INSERT_ONE => [['test' => $i]], ];
 }
 return $this->collection->bulkWrite($operations, ['writeConcern' => new WriteConcern('majority')]);
 }
 private function dropAndRecreateCollection() {
 $this->client->selectCollection($this->getDatabaseName(), $this->getCollectionName())->drop();
 $this->client->selectDatabase($this->getDatabaseName())->command(['create' => $this->getCollectionName()]);
 }
 private function getTotalConnectionsCreated(Server $server = null) {
 $server = $server ?: $this->client->getManager()->selectServer(new ReadPreference('primary'));
 $cursor = $server->executeCommand( $this->getDatabaseName(), new Command(['serverStatus' => 1]), new ReadPreference(ReadPreference::RP_PRIMARY) );
 $cursor->setTypeMap(['root' => 'array', 'document' => 'array']);
 $document = current($cursor->toArray());
 if (isset($document['connections'], $document['connections']['totalCreated'])) {
 return (int) $document['connections']['totalCreated'];
 }
 throw new UnexpectedValueException('Could not determine number of total connections');
 }
 private function waitForMasterReelection() {
 try {
 $this->insertDocuments(1);
 return;
 }
 catch (DriverException $e) {
 $this->client->getManager()->selectServer(new ReadPreference('primary'));
 return;
 }
 $this->fail('Expected primary to be re-elected within 20 seconds.');
 }
 }
 