Skip to content

259 As a developer, I want the connection pool as an instance #263

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion example/package-lock.json

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

15 changes: 6 additions & 9 deletions src/client.ts
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,7 @@ export class Client {
private publishers = new Map<string, PublisherMappedValue>()
private compressions = new Map<CompressionType, Compression>()
private connection: Connection
private pool: ConnectionPool = new ConnectionPool()

private constructor(
private readonly logger: Logger,
Expand Down Expand Up @@ -115,7 +116,7 @@ export class Client {

private async closeConnectionIfUnused(connection: Connection, params: ClosingParams) {
if (connection.refCount <= 0) {
ConnectionPool.removeCachedConnection(this.connection)
this.pool.removeCachedConnection(this.connection)
await this.connection.close({ ...params, manuallyClose: true })
}
}
Expand Down Expand Up @@ -162,7 +163,7 @@ export class Client {
publisherRef: streamPublisherParams.publisherRef,
})
}
const publisher = new StreamPublisher(streamPublisherParams, lastPublishingId, filter)
const publisher = new StreamPublisher(this.pool, streamPublisherParams, lastPublishingId, filter)
connection.registerForClosePublisher(publisher.extendedId, params.stream, async () => {
await publisher.close(false)
this.publishers.delete(publisher.extendedId)
Expand Down Expand Up @@ -203,6 +204,7 @@ export class Client {
}

const consumer = new StreamConsumer(
this.pool,
handle,
{
connection,
Expand Down Expand Up @@ -631,12 +633,7 @@ export class Client {
if (!chosenNode) {
throw new Error(`Stream was not found on any node`)
}
const cachedConnection = ConnectionPool.getUsableCachedConnection(
purpose,
streamName,
this.connection.vhost,
chosenNode.host
)
const cachedConnection = this.pool.getCachedConnection(purpose, streamName, this.connection.vhost, chosenNode.host)
if (cachedConnection) return cachedConnection

const newConnection = await this.getConnectionOnChosenNode(
Expand All @@ -647,7 +644,7 @@ export class Client {
connectionClosedListener
)

ConnectionPool.cacheConnection(purpose, streamName, this.connection.vhost, newConnection.hostname, newConnection)
this.pool.cacheConnection(purpose, streamName, this.connection.vhost, newConnection.hostname, newConnection)
return newConnection
}

Expand Down
30 changes: 14 additions & 16 deletions src/connection_pool.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,55 +5,53 @@ type InstanceKey = string
export type ConnectionPurpose = "consumer" | "publisher"

export class ConnectionPool {
private static consumerConnectionProxies = new Map<InstanceKey, Connection[]>()
private static publisherConnectionProxies = new Map<InstanceKey, Connection[]>()
private consumerConnectionProxies: Map<InstanceKey, Connection[]> = new Map<InstanceKey, Connection[]>()
private publisherConnectionProxies: Map<InstanceKey, Connection[]> = new Map<InstanceKey, Connection[]>()

public static getUsableCachedConnection(purpose: ConnectionPurpose, streamName: string, vhost: string, host: string) {
const map =
purpose === "publisher" ? ConnectionPool.publisherConnectionProxies : ConnectionPool.consumerConnectionProxies
const key = ConnectionPool.getCacheKey(streamName, vhost, host)
public getCachedConnection(purpose: ConnectionPurpose, streamName: string, vhost: string, host: string) {
const map = purpose === "publisher" ? this.publisherConnectionProxies : this.consumerConnectionProxies
const key = this.getCacheKey(streamName, vhost, host)
const proxies = map.get(key) || []
const connection = proxies.at(-1)
const refCount = connection?.refCount
return refCount !== undefined && refCount < getMaxSharedConnectionInstances() ? connection : undefined
}

public static cacheConnection(
public cacheConnection(
purpose: ConnectionPurpose,
streamName: string,
vhost: string,
host: string,
client: Connection
) {
const map =
purpose === "publisher" ? ConnectionPool.publisherConnectionProxies : ConnectionPool.consumerConnectionProxies
const key = ConnectionPool.getCacheKey(streamName, vhost, host)
const map = purpose === "publisher" ? this.publisherConnectionProxies : this.consumerConnectionProxies
const key = this.getCacheKey(streamName, vhost, host)
const currentlyCached = map.get(key) || []
currentlyCached.push(client)
map.set(key, currentlyCached)
}

public static removeIfUnused(connection: Connection) {
public removeIfUnused(connection: Connection) {
if (connection.refCount <= 0) {
ConnectionPool.removeCachedConnection(connection)
this.removeCachedConnection(connection)
return true
}
return false
}

public static removeCachedConnection(connection: Connection) {
public removeCachedConnection(connection: Connection) {
const { leader, streamName, hostname: host, vhost } = connection
if (streamName === undefined) return
const m = leader ? ConnectionPool.publisherConnectionProxies : ConnectionPool.consumerConnectionProxies
const k = ConnectionPool.getCacheKey(streamName, vhost, host)
const m = leader ? this.publisherConnectionProxies : this.consumerConnectionProxies
const k = this.getCacheKey(streamName, vhost, host)
const mappedClientList = m.get(k)
if (mappedClientList) {
const filtered = mappedClientList.filter((c) => c !== connection)
m.set(k, filtered)
}
}

private static getCacheKey(streamName: string, vhost: string, host: string) {
private getCacheKey(streamName: string, vhost: string, host: string) {
return `${streamName}@${vhost}@${host}`
}
}
3 changes: 2 additions & 1 deletion src/consumer.ts
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,7 @@ export class StreamConsumer implements Consumer {
private singleActive: boolean = false

constructor(
private pool: ConnectionPool,
handle: ConsumerFunc,
params: {
connection: Connection
Expand Down Expand Up @@ -99,7 +100,7 @@ export class StreamConsumer implements Consumer {
async close(manuallyClose: boolean): Promise<void> {
this.closed = true
this.connection.decrRefCount()
if (ConnectionPool.removeIfUnused(this.connection)) {
if (this.pool.removeIfUnused(this.connection)) {
await this.connection.close({ closingCode: 0, closingReason: "", manuallyClose })
}
}
Expand Down
3 changes: 2 additions & 1 deletion src/publisher.ts
Original file line number Diff line number Diff line change
Expand Up @@ -173,6 +173,7 @@ export class StreamPublisher implements Publisher {
private _closed = false

constructor(
private pool: ConnectionPool,
params: {
connection: Connection
stream: string
Expand Down Expand Up @@ -281,7 +282,7 @@ export class StreamPublisher implements Publisher {
if (!this.closed) {
await this.flush()
this.connection.decrRefCount()
if (ConnectionPool.removeIfUnused(this.connection)) {
if (this.pool.removeIfUnused(this.connection)) {
await this.connection.close({ closingCode: 0, closingReason: "", manuallyClose })
}
}
Expand Down