export class Endpoint<Name extends string = string> {
readonly [endpointTypeId] = endpointTypeId;
readonly participant: ParticipantHandle<Name>;
private readonly transport: EndpointTransport;
private readonly inbox: EndpointInbox;
private constructor(
participant: ParticipantHandle<Name>,
transport: EndpointTransport,
inbox: EndpointInbox,
) {
this.participant = participant;
this.transport = transport;
this.inbox = inbox;
}
static [endpointConstruction]<const Name extends string>(
attachment: AttachedEndpoint<Name>,
inbox: EndpointInbox,
): Endpoint<Name> {
return new Endpoint(attachment.participant, attachment.transport, inbox);
}
/**
* Observe messages delivered after this stream is subscribed. Conversation
* sockets retain their own ordered delivery queues independently.
* @returns Live endpoint delivery stream.
*/
messages(): Stream.Stream<ReceivedMessage, NetworkError> {
return this.inbox.messages;
}
/**
* Open a conversation through this endpoint's ordinary protocol attachment.
* The opener is included in the resulting address automatically.
* @param participants Nonempty addressed participant set.
* @returns A conversation socket bound to this endpoint.
*/
open(
...participants: ConversationParticipants
): Effect.Effect<ConversationSocket, NetworkError> {
const [first, ...rest] = participants;
const ids: ParticipantIds = [
first.id,
...rest.map((participant) => participant.id),
];
const addressed = addressedParticipants(this.participant, participants);
return this.transport.openConversation(ids).pipe(
Effect.flatMap((opened) =>
this.inbox.conversation(opened.conversationId).pipe(
Effect.map((messages) => ({
messages,
opened,
})),
),
),
Effect.map(({ messages, opened }) => {
const address = makeConversationAddress(
opened.conversationId,
addressed,
);
return makeConversationSocket(
this.participant,
address,
messages,
(content) => this.transport.send(address.conversationId, content),
);
}),
);
}
/**
* Bind this endpoint as the sender for an existing address.
* @param address Participant-independent conversation address.
* @returns Endpoint-bound socket when this endpoint is addressed.
*/
socket(
address: ConversationAddress,
): Effect.Effect<ConversationSocket, NetworkError> {
const isParticipant = address.participants.some(
(participant) => participant.id === this.participant.id,
);
return isParticipant
? this.inbox
.conversation(address.conversationId)
.pipe(
Effect.map((messages) =>
makeConversationSocket(
this.participant,
address,
messages,
(content) =>
this.transport.send(address.conversationId, content),
),
),
)
: Effect.fail(
networkError(
"socket",
`participant ${this.participant.name} is not addressed by the conversation`,
),
);
}
}