Encapsulate session handling from protocol to new SessionManager (#247)
This commit is contained in:
parent
a6ed01bec0
commit
ae320646bf
1 changed files with 444 additions and 200 deletions
|
|
@ -25,9 +25,8 @@ use Mcp\Schema\JsonRpc\ResultInterface;
|
|||
use Mcp\Schema\Request\InitializeRequest;
|
||||
use Mcp\Server\Handler\Notification\NotificationHandlerInterface;
|
||||
use Mcp\Server\Handler\Request\RequestHandlerInterface;
|
||||
use Mcp\Server\Session\SessionFactoryInterface;
|
||||
use Mcp\Server\Session\SessionInterface;
|
||||
use Mcp\Server\Session\SessionStoreInterface;
|
||||
use Mcp\Server\Session\SessionManagerInterface;
|
||||
use Mcp\Server\Transport\TransportInterface;
|
||||
use Psr\EventDispatcher\EventDispatcherInterface;
|
||||
use Psr\Log\LoggerInterface;
|
||||
|
|
@ -70,8 +69,7 @@ class Protocol
|
|||
private readonly array $requestHandlers,
|
||||
private readonly array $notificationHandlers,
|
||||
private readonly MessageFactory $messageFactory,
|
||||
private readonly SessionFactoryInterface $sessionFactory,
|
||||
private readonly SessionStoreInterface $sessionStore,
|
||||
private readonly SessionManagerInterface $sessionManager,
|
||||
private readonly LoggerInterface $logger = new NullLogger(),
|
||||
private readonly ?EventDispatcherInterface $eventDispatcher = null,
|
||||
) {
|
||||
|
|
@ -112,7 +110,7 @@ class Protocol
|
|||
{
|
||||
$this->logger->info('Received message to process.', ['message' => $input]);
|
||||
|
||||
$this->gcSessions();
|
||||
$this->sessionManager->gc();
|
||||
|
||||
try {
|
||||
$messages = $this->messageFactory->create($input);
|
||||
|
|
@ -410,7 +408,7 @@ class Protocol
|
|||
*/
|
||||
public function consumeOutgoingMessages(Uuid $sessionId): array
|
||||
{
|
||||
$session = $this->sessionFactory->createWithId($sessionId, $this->sessionStore);
|
||||
$session = $this->sessionManager->createWithId($sessionId);
|
||||
$queue = $session->get(self::SESSION_OUTGOING_QUEUE, []);
|
||||
$session->set(self::SESSION_OUTGOING_QUEUE, []);
|
||||
$session->save();
|
||||
|
|
@ -429,7 +427,7 @@ class Protocol
|
|||
*/
|
||||
public function checkResponse(int $requestId, Uuid $sessionId): Response|Error|null
|
||||
{
|
||||
$session = $this->sessionFactory->createWithId($sessionId, $this->sessionStore);
|
||||
$session = $this->sessionManager->createWithId($sessionId);
|
||||
$responseData = $session->get(self::SESSION_RESPONSES.".{$requestId}");
|
||||
|
||||
if (null === $responseData) {
|
||||
|
|
@ -471,7 +469,7 @@ class Protocol
|
|||
*/
|
||||
public function getPendingRequests(Uuid $sessionId): array
|
||||
{
|
||||
$session = $this->sessionFactory->createWithId($sessionId, $this->sessionStore);
|
||||
$session = $this->sessionManager->createWithId($sessionId);
|
||||
|
||||
return $session->get(self::SESSION_PENDING_REQUESTS, []);
|
||||
}
|
||||
|
|
@ -498,7 +496,7 @@ class Protocol
|
|||
return;
|
||||
}
|
||||
|
||||
$session = $this->sessionFactory->createWithId($sessionId, $this->sessionStore);
|
||||
$session = $this->sessionManager->createWithId($sessionId);
|
||||
|
||||
$payloadSessionId = $yieldedValue['session_id'] ?? null;
|
||||
if (\is_string($payloadSessionId) && $payloadSessionId !== $sessionId->toRfc4122()) {
|
||||
|
|
@ -582,7 +580,7 @@ class Protocol
|
|||
return null;
|
||||
}
|
||||
|
||||
$session = $this->sessionFactory->create($this->sessionStore);
|
||||
$session = $this->sessionManager->create();
|
||||
$this->logger->debug('Created new session for initialize', [
|
||||
'session_id' => $session->getId()->toRfc4122(),
|
||||
]);
|
||||
|
|
@ -599,33 +597,14 @@ class Protocol
|
|||
return null;
|
||||
}
|
||||
|
||||
if (!$this->sessionStore->exists($sessionId)) {
|
||||
if (!$this->sessionManager->exists($sessionId)) {
|
||||
$error = Error::forInvalidRequest('Session not found or has expired.');
|
||||
$this->sendResponse($transport, $error, null, ['status_code' => 404]);
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
return $this->sessionFactory->createWithId($sessionId, $this->sessionStore);
|
||||
}
|
||||
|
||||
/**
|
||||
* Run garbage collection on expired sessions.
|
||||
* Uses the session store's internal TTL configuration.
|
||||
*/
|
||||
private function gcSessions(): void
|
||||
{
|
||||
if (random_int(0, 100) > 1) {
|
||||
return;
|
||||
}
|
||||
|
||||
$deletedSessions = $this->sessionStore->gc();
|
||||
if (!empty($deletedSessions)) {
|
||||
$this->logger->debug('Garbage collected expired sessions.', [
|
||||
'count' => \count($deletedSessions),
|
||||
'session_ids' => array_map(static fn (Uuid $id) => $id->toRfc4122(), $deletedSessions),
|
||||
]);
|
||||
}
|
||||
return $this->sessionManager->createWithId($sessionId);
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -633,7 +612,7 @@ class Protocol
|
|||
*/
|
||||
public function destroySession(Uuid $sessionId): void
|
||||
{
|
||||
$this->sessionStore->destroy($sessionId);
|
||||
$this->sessionManager->destroy($sessionId);
|
||||
$this->logger->info('Session destroyed.', ['session_id' => $sessionId->toRfc4122()]);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue