NServiceBus.ServiceFabricHost

NServiceBus.ServiceFabricHost.cs

// wrap ASF Stateless Service.  Customer would implement this base class instead of StatelessService 
public abstract class NServiceBusStatelessService : StatelessService, IProvideEndPointConfiguration
{
  //plumbing for starting NSB on Service start
  protected override IEnumerable<ServiceInstanceListener> CreateServiceInstanceListeners()
  {
    yield return new ServiceInstanceListener(
      createCommunicationListener:
        _ => new NServiceBusCommunicationsListener(this),
      name:
        nameof(NServiceBusCommunicationsListener);
  }

  protected abstract Task<EndpointConfiguration> BuildEndpointConfigurationAsync(CancellationToken cancelToken);

  //explicit implementation to keep member permissions nice
  Task<EndpointConfiguration> IProvideEndPointConfiguration.BuildEndpointConfigurationAsync(CancellationToken cancelToken)
  {
    return BuildEndpointConfigurationAsync(cancelToken);
  }
}

// wrap ASF Stateful Service.  Customer would implement this base class instead of StatefulService 
public abstract class NServiceBusStatefulService : StatefulService, IProvideEndPointConfiguration
{
  //plumbing for starting NSB on Service start
  protected override IEnumerable<ServiceReplicaListener> CreateServiceReplicaListeners()
  {
    yield return new ServiceReplicaListener(
      createCommunicationListener:
        _ => new NServiceBusCommunicationsListener(this),
      name:
        nameof(NServiceBusCommunicationsListener),
      listenOnSecondary:
        false);
  }
  
  // implementor responsible for Bind<IReliableStateManager>().ToConstant(this.StateManager) (and plugging that in to
  // EndpointConfiguration.UseContainer) to allow handlers direct access to Reliable State
  protected abstract Task<EndpointConfiguration> BuildEndpointConfigurationAsync(CancellationToken cancelToken);

  //explicit implementation to keep member permissions nice
  Task<EndpointConfiguration> IProvideEndPointConfiguration.BuildEndpointConfigurationAsync(CancellationToken cancelToken)
  {
    return BuildEndpointConfigurationAsync(cancelToken);
  }
}

// bridge interface for NServiceBusCommunicationsListener
internal interface IProvideEndPointConfiguration
{  
  Task<EndpointConfiguration> BuildEndpointConfigurationAsync(CancellationToken cancelToken);
}

public class NServiceBusCommunicationsListener : ICommunicationListener // asf interface
{ 
  pirvate IProvideEndPointConfiguration _service;
  private IEndpointInstance _endpoint;

  public NServiceBusCommunicationsListener(IProvideEndPointConfiguration  service){_service = service;}

  public async Task<string> OpenAsync(CancellationToken cancellationToken)
  {
    var endpointConfig = await _service.BuildEndpointConfigurationAsync(cancellationToken); 
    _endpoint = await ((await Endpoint.Create(endpointConfig)).Start();

    return endpointConfig.GetSettings().Get<EndpointName>();
  }
  
  public async Task CloseAsync(CancellationToken cancellationToken){ _endpoint?.Stop();  }
  public void Abort(){  _endpoint?.Stop(); }         
}