remove rpcService.onBeforeSend

Author: maxenglander

.gitignore

@@ -1,6 +1,7 @@
 NOTES.txt
 .nvmrc
 *.swp
+*.txt
 build/
 dist/
 node_modules/

src/rpc/service.ts

@@ -12,13 +12,11 @@ import { ITransport } from '../transport';
 
 // Encodes, sends, decodes and receives RPC messages.
 export class RpcService implements IRpcService {
-  private beforeSendHooks: Set<RpcBeforeSendHook>;
   private codec: ICodec;
   private receivers: Set<RpcReceiver>;
   private transport: ITransport;
 
   constructor(options: IRpcServiceOptions = {}) {
-    this.beforeSendHooks = new Set<RpcBeforeSendHook>();
     this.codec = options.codec;
     this.receivers = new Set<RpcReceiver>();
     this.transport = options.transport;
@@ -48,16 +46,6 @@ export class RpcService implements IRpcService {
     }
   }
 
-  // Register a hook to be invoked before a message is sent.
-  public onBeforeSend(hook: RpcBeforeSendHook): IDetacher {
-    this.beforeSendHooks.add(hook);
-    return {
-      detach: (): void => {
-        this.beforeSendHooks.delete(hook);
-      }
-    };
-  }
-
   // Register a receiver for messages of any procedure and call type.
   public onReceive(receiver: RpcReceiver): IDetacher {
     this.receivers.add(receiver);
@@ -73,9 +61,6 @@ export class RpcService implements IRpcService {
     endpoint: IEndpoint,
     message: IRpcMessage
   ): Promise<void> {
-    for (const hook of this.beforeSendHooks) {
-      await hook(endpoint, message);
-    }
     const encoded = this.codec.encode(message);
     return await this.transport.send(endpoint, encoded);
   }

src/rpc/types.ts

@@ -16,7 +16,6 @@ export type RpcReceiver = (endpoint: IEndpoint, message: IRpcMessage) => any;
 export interface IRpcService {
   close(): Promise<void>;
   listen(endpoint: IEndpoint): Promise<void>;
-  onBeforeSend(hook: RpcBeforeSendHook): IDetacher;
   onReceive(receiver: RpcReceiver): IDetacher;
   send(
     endpoint: IEndpoint,

src/server.ts

@@ -34,7 +34,7 @@ export class Server implements IServer {
   private lastApplied: number;
   public readonly log: ILog;
   public readonly logger: ILogger;
-  public readonly rpcService: IRpcService;
+  private readonly rpcService: IRpcService;
   private rpcServiceDetachers: Set<IDetacher>;
   private state: IState;
   public readonly stateMachine: IStateMachine;
@@ -119,6 +119,17 @@ export class Server implements IServer {
     await this.state.handleRpcMessage(endpoint, message);
   }
 
+  public async sendRpcMessage(endpoint: IEndpoint, message: IRpcMessage): Promise<void> {
+    // Before responding to an RPC request, the recipient `Server`
+    // updates persistent state on stable storage.
+    // > *§5. "...(Updated on stable storage before responding)..."*
+    // > *§5.6 "...Raft’s RPCs...require the recipient to persist..."*
+    if (isRpcResponse(message)) {
+      await this.updatePersistentState();
+    }
+    await this.rpcService.send(endpoint, message);
+  }
+
   // When the term is updated, it is not immediately
   // persisted because, as the Raft paper says, the
   // term is part of persistent state that is:
@@ -155,22 +166,13 @@ export class Server implements IServer {
     this.commitIndex = this.lastApplied = 0;
 
     this.logger.debug('Loading persistent state');
+
     // The current term is...
     // > *§5. "...initialized to zero on first boot..."*
     await this.currentTerm.readIfExistsElseSetAndWrite(0)
 
     this.logger.debug('Starting RPC service');
 
-    this.rpcServiceDetachers.add(this.rpcService.onBeforeSend(async (endpoint, message) => {
-      // Before responding to an RPC request, the recipient `Server`
-      // updates persistent state on stable storage.
-      // > *§5. "...(Updated on stable storage before responding)..."*
-      // > *§5.6 "...Raft’s RPCs...require the recipient to persist..."*
-      if (isRpcResponse(message)) {
-        await this.updatePersistentState();
-      }
-    }));
-
     this.rpcServiceDetachers.add(this.rpcService.onReceive(async (endpoint, message) => {
       await this.handleRpcMessage(endpoint, message);
     }));

src/state/candidate.ts

@@ -108,7 +108,7 @@ export class CandidateState implements IState {
     const lastLogIndex = this.server.log.getLastIndex();
 
     for (const serverEndpoint of this.server.getServerEndpoints()) {
-      this.server.rpcService.send(
+      this.server.sendRpcMessage(
         serverEndpoint,
         createRequestVoteRpcRequest({
           candidateId: this.server.id,

src/state/follower.ts

@@ -96,7 +96,7 @@ export class FollowerState implements IState {
       }
     }
 
-    await this.server.rpcService.send(
+    await this.server.sendRpcMessage(
       endpoint,
       createAppendEntriesRpcResponse({
         // The followerCommit field is not part of the Raft spec. It is a
@@ -155,7 +155,7 @@ export class FollowerState implements IState {
       );
     }
 
-    await this.server.rpcService.send(
+    await this.server.sendRpcMessage(
       endpoint,
       createRequestVoteRpcResponse({
         term: currentTerm,

src/state/leader.ts

@@ -84,7 +84,7 @@ export class LeaderState implements IState {
     for (const serverId of this.server.getServerIds()) {
       if (this.server.log.getLastIndex() >= this.nextIndex[serverId]) {
         const serverEndpoint = this.server.getCluster().servers[serverId];
-        this.server.rpcService.send(
+        this.server.sendRpcMessage(
           serverEndpoint,
           createAppendEntriesRpcRequest({
             entries: this.server.log.slice(this.nextIndex[serverId]),
@@ -103,7 +103,7 @@ export class LeaderState implements IState {
 
   private sendHeartbeats() {
     for (const serverEndpoint of this.server.getServerEndpoints()) {
-      this.server.rpcService.send(
+      this.server.sendRpcMessage(
         serverEndpoint,
         createAppendEntriesRpcRequest({
           entries: [],

src/types.ts

@@ -25,7 +25,6 @@ export interface IServer {
   readonly id: ServerId;
   readonly log: ILog;
   readonly logger: ILogger;
-  readonly rpcService: IRpcService;
   readonly stateMachine: IStateMachine;
   getCommitIndex(): number;
   getCluster(): ICluster;
@@ -36,6 +35,7 @@ export interface IServer {
   getState(): IState;
   getVotedFor(): ServerId;
   handleClientRequest(request: IClientRequest): Promise<IClientResponse>;
+  sendRpcMessage(endpoint: IEndpoint, message: IRpcMessage): Promise<void>;
   setCurrentTerm(newTerm: number): void;
   setVotedFor(candidateId: ServerId): void;
   start(): Promise<void>;

test/server.test.ts

@@ -145,7 +145,7 @@ describe('server', function() {
         spyLogWrite = sinon.spy(log, "write");
         spyTransportSend = sinon.spy(transport, "send");
         spyVotedForWrite = sinon.spy(votedFor, "write");
-        return server.rpcService.send(peerEndpoint, message);
+        return server.sendRpcMessage(peerEndpoint, message);
       });
 
       if (isRpcRequest(message)) {