<commit_before>/* DO NOT EDIT: File is auto-generated */
package <%= to_package(repository.imit.qualified_session_manager_name) %>;

@javax.annotation.Generated( "Domgen" )
@java.lang.SuppressWarnings( { "unchecked", "PMD.UnusedPrivateField", "PMD.CollapsibleIfStatements", "PMD.UnusedFormalParameter" } )
<% if repository.ee.use_cdi? -%>
@javax.enterprise.context.ApplicationScoped
@javax.transaction.Transactional( javax.transaction.Transactional.TxType.REQUIRED )
@javax.enterprise.inject.Typed( { <%= repository.imit.qualified_poll_service_name %>.class, org.realityforge.replicant.server.EntityMessageEndpoint.class, org.realityforge.ssf.SessionManager.class, org.realityforge.replicant.server.transport.ReplicantSessionManager.class, <%= repository.service_by_name(repository.imit.subscription_manager).ejb.qualified_service_name %>.class } )
<% else -%>
@javax.ejb.Singleton( name = <%= repository.service_by_name(repository.imit.subscription_manager).ejb.qualified_service_name %>.NAME )
@javax.ejb.ConcurrencyManagement( javax.ejb.ConcurrencyManagementType.BEAN )
@javax.ejb.Local( { <%= repository.imit.qualified_poll_service_name %>.class, org.realityforge.replicant.server.EntityMessageEndpoint.class, org.realityforge.ssf.SessionManager.class, org.realityforge.replicant.server.transport.ReplicantSessionManager.class, <%= repository.service_by_name(repository.imit.subscription_manager).ejb.qualified_service_name %>.class } )
<% end -%>
public class <%= repository.imit.session_manager_name %>
  extends org.realityforge.replicant.server.transport.ReplicantSessionManagerImpl
  implements <%= repository.service_by_name(repository.imit.subscription_manager).ejb.qualified_service_name %><% if repository.imit.poll_replicate_mode? %>, <%= repository.imit.qualified_poll_service_name %><% end %>
{
<% repository.imit.graphs.each_with_index do |graph, i| -%>
  private static final org.realityforge.replicant.server.transport.ChannelMetaData $CHANNEL_<%= i %> = new org.realityforge.replicant.server.transport.ChannelMetaData( <%= repository.imit.qualified_graph_enum_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>.ordinal(), "<%= graph.name %>", <%= !graph.instance_root? %>, org.realityforge.replicant.server.transport.ChannelMetaData.FilterType.<%= graph.filter_parameter.nil? ? 'NONE' : graph.filter_parameter.immutable? ? 'STATIC' : 'DYNAMIC' %>, <%= graph.cacheable? %> );
<% end -%>
  private static final org.realityforge.replicant.server.transport.ChannelMetaData[] CHANNELS = new org.realityforge.replicant.server.transport.ChannelMetaData[]{<% repository.imit.graphs.each_with_index do |graph, i| -%> $CHANNEL_<%= i %>,<% end %> };
  private static final java.util.logging.Logger LOG = java.util.logging.Logger.getLogger( <%= repository.imit.session_manager_name %>.class.getName() );
  @javax.inject.Inject
  private <%= repository.imit.qualified_graph_encoder_name %> _encoder;
<%
entities = repository.imit.graphs.select{|g| g.instance_root?}.collect{|g|repository.entity_by_name(g.instance_root)}
repository.imit.graphs.select{|g| !g.instance_root?}.each do |graph|
  graph.type_roots.each do |type_root_name|
    entities << repository.entity_by_name(type_root_name)
  end
end

entities.uniq.each do |entity| -%>
  @javax.inject.Inject
  private <%= entity.dao.jpa.qualified_dao_service_name %> _<%= Reality::Naming.camelize("#{entity.data_module.name}#{entity.name}") %>Repository;
<% end -%>
  @javax.inject.Inject
  private <%= repository.service_by_name(repository.imit.session_context_service).ejb.qualified_service_name %> _context;
  @javax.annotation.Resource
  private javax.transaction.TransactionSynchronizationRegistry _registry;

  @javax.annotation.Nonnull
  @java.lang.Override
  protected javax.transaction.TransactionSynchronizationRegistry getRegistry()
  {
    return _registry;
  }

  @javax.annotation.Nonnull
  @java.lang.Override
  protected org.realityforge.replicant.server.transport.ChannelMetaData[] getChannelMetaData()
  {
    return CHANNELS;
  }

  /**
   * Remove idle session changes. This is called by a scheduler.
   */
  @java.lang.Override
  public void removeIdleSessions()
  {
    final int maxIdleTime = 1000 * 60 * 5;
    final int removedSessions = removeIdleSessions( maxIdleTime );
    if ( 0 != removedSessions && LOG.isLoggable( java.util.logging.Level.INFO ) )
    {
      LOG.info( "Removed " + removedSessions + " idle sessions" );
    }
  }

  @java.lang.Override
  protected void processDeleteMessages( @javax.annotation.Nonnull final org.realityforge.replicant.server.EntityMessage message, @javax.annotation.Nonnull final java.util.Collection<org.realityforge.replicant.server.transport.ReplicantSession> sessions )
  {
    final java.util.Map<String, java.io.Serializable> routingKeys = message.getRoutingKeys();
<%
  repository.imit.graphs.select{|graph| graph.instance_root?}.each do |graph|

    instance_root = repository.entity_by_name(graph.instance_root)
    type = instance_root.primary_key.jpa.non_primitive_java_type
    %>
    final <%= type %> <%= graph.name %> = (<%= type %>) routingKeys.get( <%= repository.imit.qualified_router_impl_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>_KEY );
    if ( null != <%= graph.name %> )
    {
      final org.realityforge.replicant.server.ChannelDescriptor descriptor = new org.realityforge.replicant.server.ChannelDescriptor( <%= repository.imit.qualified_graph_enum_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>.ordinal(), <%= graph.instance_root? ? graph.name : 'null' %> );
<%
  extra_interest_filter = ''
  if graph.filtered?
    graph.routing_keys.each do |routing_key|
      var_name = Reality::Naming.camelize(routing_key.name)
      extra_interest_filter << ", #{var_name}"
      param_type = routing_key.target_attribute.jpa.non_primitive_java_type
      param_type = "java.util.List<#{param_type}>" if routing_key.multivalued?
-%>
      final <%= param_type %> <%= var_name %> = (<%= param_type %>) routingKeys.get( <%= repository.imit.qualified_router_impl_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>_<%= Reality::Naming.uppercase_constantize(routing_key.name) %>_KEY );
<%
    end
  end

  customize_filter = ", session#{graph.instance_root? ? ", #{graph.name}" : ''}"
  customize_filter = "#{customize_filter}, (#{Domgen::Java.java_type(graph.filter_parameter, :ee, :default)}) entry.getFilter()" if graph.filter_parameter?
  customize_filter = "#{customize_filter}#{extra_interest_filter}"
-%>
      for ( final org.realityforge.replicant.server.transport.ReplicantSession session : sessions )
      {
        final org.realityforge.replicant.server.transport.SubscriptionEntry entry = session.findSubscriptionEntry( descriptor );
        if ( null != entry )
        {
          final org.realityforge.replicant.server.EntityMessage m = _context.filterMessageOfInterestIn<%= graph.name %>Graph( message<%= customize_filter %> );
          if( null != m )
          {
            if( m.isDelete() && <%= repository.imit.qualified_message_constants_name %>.<%= Reality::Naming.uppercase_constantize(instance_root.data_module.name.to_s) %>_<%= Reality::Naming.uppercase_constantize(instance_root.name.to_s) %> == m.getTypeID() )
            {
              performUnsubscribe( session, entry, true );
            }
          }
        }
      }
    }
<% end -%>
  }

  @java.lang.Override
  protected void processUpdateMessages( @javax.annotation.Nonnull final org.realityforge.replicant.server.EntityMessage message, @javax.annotation.Nonnull final java.util.Collection<org.realityforge.replicant.server.transport.ReplicantSession> sessions, @javax.annotation.Nonnull org.realityforge.replicant.server.ChangeAccumulator accumulator )
  {
    final java.util.Map<String, java.io.Serializable> routingKeys = message.getRoutingKeys();
<%
  repository.imit.graphs.each do |graph|

  type = 'Boolean'
  if graph.instance_root?
    instance_root = repository.entity_by_name(graph.instance_root)
    type = instance_root.primary_key.jpa.non_primitive_java_type
  end
-%>
    final <%= type %> <%= graph.name %> = (<%= type %>) routingKeys.get( <%= repository.imit.qualified_router_impl_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>_KEY );
    if ( null != <%= graph.name %> )
    {
      final org.realityforge.replicant.server.ChannelDescriptor descriptor = new org.realityforge.replicant.server.ChannelDescriptor( <%= repository.imit.qualified_graph_enum_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>.ordinal(), <%= graph.instance_root? ? graph.name : 'null' %> );
<%
  extra_interest_filter = ''
  if graph.filtered?
    graph.routing_keys.each do |routing_key|
      var_name = Reality::Naming.camelize(routing_key.name)
      extra_interest_filter << ", #{var_name}"
      param_type = routing_key.target_attribute.jpa.non_primitive_java_type
      param_type = "java.util.List<#{param_type}>" if routing_key.multivalued?
-%>
      final <%= param_type %> <%= var_name %> = (<%= param_type %>) routingKeys.get( <%= repository.imit.qualified_router_impl_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>_<%= Reality::Naming.uppercase_constantize(routing_key.name) %>_KEY );
<%
    end
  end

  customize_filter = ", session#{graph.instance_root? ? ", #{graph.name}" : ''}"
  customize_filter = "#{customize_filter}, (#{Domgen::Java.java_type(graph.filter_parameter, :ee, :default)}) entry.getFilter()" if graph.filter_parameter?
  customize_filter = "#{customize_filter}#{extra_interest_filter}"
-%>
<% if graph.cacheable? && !graph.external_cache_management? -%>
      deleteCacheEntry( descriptor );
<% end -%>
      for ( final org.realityforge.replicant.server.transport.ReplicantSession session : sessions )
      {
        final org.realityforge.replicant.server.transport.SubscriptionEntry entry = session.findSubscriptionEntry( descriptor );
        if ( null != entry )
        {
          final org.realityforge.replicant.server.EntityMessage m = _context.filterMessageOfInterestIn<%= graph.name %>Graph( message<%= customize_filter %> );
          if( null != m )
          {
            accumulator.addChange( session, new org.realityforge.replicant.server.Change( m, descriptor.getChannelID(), descriptor.getSubChannelID() ) );
          }
        }
      }
    }
<% end -%>
  }

<%
  repository.imit.graphs.each do |graph|

  param = []
  param_value = []
  pk_accessor = ''
  if graph.instance_root?
    instance_root = repository.entity_by_name(graph.instance_root)
    param << "@javax.annotation.Nonnull final #{instance_root.jpa.qualified_name} entity"
    param_value << 'entity'
    pk_accessor = "entity.#{getter_for(instance_root.primary_key)}"
  end
  update_param = param.dup
  update_param_value = param_value.dup
  filter_param = nil
  if graph.filter_parameter
    p = graph.filter_parameter
    filter_param = "#{nullability_annotation(p.nullable?)} #{Domgen::Java.java_type(p, :ee, :default)} filter"
    update_param << filter_param
    update_param_value << 'filter'
  end
  param = param.join(', ')
  update_param = update_param.join(', ')

  param = ", #{param}" if param.size > 0
  update_param = ", #{update_param}" if update_param.size > 0
-%>

  @java.lang.Override
  public <%= graph.cacheable? ? 'boolean': 'void' %> subscribeTo<%= graph.name %>( @javax.annotation.Nonnull final String sessionID<%= update_param %><% if graph.cacheable? -%>, @javax.annotation.Nullable final String cacheKey<% end %> )
    throws <%= repository.exception_by_name(repository.imit.invalid_session_exception).ee.qualified_name %>
  {
    <%= graph.cacheable? ? 'final CacheStatus status = ': '' %>subscribe( sessionID, new org.realityforge.replicant.server.ChannelDescriptor( <%= repository.imit.qualified_graph_enum_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>.ordinal(), <%= graph.instance_root? ? "#{pk_accessor}" : 'null' %> ), <%= graph.filter_parameter.nil? ? 'null' : 'filter' %>, <%= graph.cacheable? ? 'cacheKey' : 'null' %> );
<% if graph.cacheable? -%>
    return status != CacheStatus.USE;
<% end -%>
  }
<% if graph.instance_root?

  root_type = Domgen::Java.java_type(instance_root.primary_key, :ee, :default)
-%>

  public void subscribeToMultiple<%= Reality::Naming.pluralize(graph.name) %>( @javax.annotation.Nonnull final String sessionID, @javax.annotation.Nonnull final java.util.List<<%= root_type %>> roots<%= filter_param.nil? ? '' : ", #{filter_param}" %> )
    throws <%= repository.exception_by_name(repository.imit.invalid_session_exception).ee.qualified_name %>
  {
    bulkSubscribe( sessionID, <%= repository.imit.qualified_graph_enum_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>.ordinal(), (java.util.Collection<java.io.Serializable>) (java.util.Collection) roots, <%= filter_param.nil? ? 'null' : 'filter' %>, true );
  }

  public void unsubscribeFromMultiple<%= Reality::Naming.pluralize(graph.name) %>( @javax.annotation.Nonnull final String sessionID, @javax.annotation.Nonnull final java.util.List<<%= root_type %>> roots )
    throws <%= repository.exception_by_name(repository.imit.invalid_session_exception).ee.qualified_name %>
  {
    bulkUnsubscribe( sessionID, <%= repository.imit.qualified_graph_enum_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>.ordinal(), (java.util.Collection<java.io.Serializable>) (java.util.Collection) roots, true );
  }
<% end -%>
<% if !graph.instance_root? && !graph.external_data_load? -%>
  private void collect<%= graph.name %>( @javax.annotation.Nonnull org.realityforge.replicant.server.EntityMessageSet messages )
  {
    // Add to a list so that when routing we limit the number of new entities that are loaded during routing
    final java.util.LinkedList<Object> objects = new java.util.LinkedList<>();
<% graph.type_roots.each do |type_root_name|
  type = repository.entity_by_name(type_root_name)
-%>
    objects.addAll( _<%= Reality::Naming.camelize("#{type.data_module.name}#{type.name}") %>Repository.findAll() );
<% end -%>
    _encoder.encodeObjects( messages, objects );
  }

<% end -%>
<% if graph.instance_root? -%>

  private org.realityforge.replicant.server.EntityMessageSet encode<%= graph.name %>( @javax.annotation.Nonnull final <%= repository.entity_by_name(graph.instance_root).jpa.qualified_name %> entity<%= graph.filter_parameter.nil? ? '' : ", #{Domgen::Java.java_type(graph.filter_parameter, :ee, :default)} filter" %> )
  {
    final org.realityforge.replicant.server.EntityMessageSet messages = new org.realityforge.replicant.server.EntityMessageSet();
    _encoder.encode<%= graph.name %>( messages, entity<%= graph.filter_parameter.nil? ? '' : ", filter" %> );
    return messages;
  }
<% end -%>
<% if graph.filter_parameter && !graph.filter_parameter.immutable? -%>
<% if graph.instance_root? -%>
  @java.lang.Override
  public void updateMultiple<%= graph.name %>Subscriptions( @javax.annotation.Nonnull final String sessionID, @javax.annotation.Nonnull final java.util.List<<%= root_type %>> roots, <%= filter_param %> )
    throws <%= repository.exception_by_name(repository.imit.invalid_session_exception).ee.qualified_name %>
  {
    bulkUpdateSubscription( sessionID, <%= repository.imit.qualified_graph_enum_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>.ordinal(), (java.util.Collection<java.io.Serializable>) (java.util.Collection) roots, filter );
<%
  processed = []
  graph.outward_graph_links.select{|graph_link| graph_link.auto?}.each do |graph_link|
     target_graph = repository.imit.graph_by_name(graph_link.target_graph)
     next unless target_graph.filtered?
     key = "#{graph_link.source_graph}=>#{graph_link.target_graph}"
     next if processed.include?(key)
     processed << key
     instance_root = repository.entity_by_name(target_graph.instance_root)
     entity = repository.entity_by_name(graph.instance_root)
-%>
    for ( final <%= root_type %> id : roots )
    {
      final <%= entity.jpa.qualified_name %> entity = _<%= Reality::Naming.camelize("#{entity.data_module.name}#{entity.name}") %>Repository.getBy<%= entity.primary_key.name %>( id );
      //Propogate the filter from <%= graph.name %> to downstream graph <%= target_graph.name %>
      for ( final <%= instance_root.jpa.qualified_name %> downstreamEntity : _context.getLinksToUpdateFor<%= graph_link.source_graph %>To<%= target_graph.name %>( entity, filter ) )
      {
        subscribeTo<%= target_graph.name %>( sessionID, downstreamEntity, filter );
      }
    }
<% end -%>
  }
<% end -%>

  @java.lang.Override
  public void update<%= graph.name %>Subscription( @javax.annotation.Nonnull final String sessionID<%= update_param %> )
    throws <%= repository.exception_by_name(repository.imit.invalid_session_exception).ee.qualified_name %>
  {
    updateSubscription( sessionID, new org.realityforge.replicant.server.ChannelDescriptor( <%= repository.imit.qualified_graph_enum_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>.ordinal(), <%= graph.instance_root? ? pk_accessor : 'null' %> ), filter );
<%
  processed = []
  graph.outward_graph_links.select{|graph_link| graph_link.auto?}.each do |graph_link|
     target_graph = repository.imit.graph_by_name(graph_link.target_graph)
     next unless target_graph.filtered?
     key = "#{graph_link.source_graph}=>#{graph_link.target_graph}"
      next if processed.include?(key)
      processed << key
      instance_root = repository.entity_by_name(target_graph.instance_root)
-%>
    //Propogate the filter from <%= graph.name %> to downstream graph <%= target_graph.name %>
    for ( final <%= instance_root.jpa.qualified_name %> downstreamEntity : _context.getLinksToUpdateFor<%= graph_link.source_graph %>To<%= target_graph.name %>( entity, filter ) )
    {
      subscribeTo<%= target_graph.name %>( sessionID, downstreamEntity, filter );
    }
<% end if graph.instance_root? -%>
  }

  protected void collectDataFor<%= graph.name %>SubscriptionUpdate( @javax.annotation.Nonnull final org.realityforge.replicant.server.transport.ReplicantSession session, @javax.annotation.Nonnull final org.realityforge.replicant.server.ChannelDescriptor descriptor, @javax.annotation.Nonnull final org.realityforge.replicant.server.ChangeSet changeSet, final <%= Domgen::Java.java_type(graph.filter_parameter, :ee, :default) %> originalFilter, final <%= Domgen::Java.java_type(graph.filter_parameter, :ee, :default) %> filter )
  {
    final org.realityforge.replicant.server.EntityMessageSet messages = new org.realityforge.replicant.server.EntityMessageSet();
<%
    if graph.instance_root?
      entity = repository.entity_by_name(graph.instance_root)
-%>
    final <%= entity.primary_key.jpa.java_type %> id = (<%= entity.primary_key.jpa.java_type %>) descriptor.getSubChannelID();
    final <%= entity.jpa.qualified_name %> entity = _<%= Reality::Naming.camelize("#{entity.data_module.name}#{entity.name}") %>Repository.getBy<%= entity.primary_key.name %>( id );
<% end %>
    _context.collectForFilterChange<%= graph.name %>( messages, <%= graph.instance_root? ? 'entity, ' : '' %>originalFilter, filter );
    changeSet.mergeAll( org.realityforge.replicant.server.ChangeUtil.toChanges( messages.getEntityMessages(), descriptor.getChannelID(), descriptor.getSubChannelID() ) );
  }
<% end -%>

  protected <% if graph.cacheable? -%>java.lang.String<% else -%>void<% end -%> collectDataFor<%= graph.name %>( @javax.annotation.<% if graph.cacheable? -%>Nullable<% else %>Nonnull<% end %> final org.realityforge.replicant.server.transport.ReplicantSession session, @javax.annotation.Nonnull final org.realityforge.replicant.server.ChannelDescriptor descriptor, @javax.annotation.Nonnull final org.realityforge.replicant.server.ChangeSet changeSet<% if graph.filter_parameter -%>, final <%= Domgen::Java.java_type(graph.filter_parameter, :ee, :default) %> filter<% end %>)
  {
<%
    if graph.instance_root?
      entity = repository.entity_by_name(graph.instance_root)
-%>
    final org.realityforge.replicant.server.EntityMessageSet messages = new org.realityforge.replicant.server.EntityMessageSet();
    final <%= entity.primary_key.jpa.java_type %> id = (<%= entity.primary_key.jpa.java_type %>) descriptor.getSubChannelID();
    final <%= entity.jpa.qualified_name %> entity = _<%= Reality::Naming.camelize("#{entity.data_module.name}#{entity.name}") %>Repository.getBy<%= entity.primary_key.name %>( id );
    _encoder.encode<%= graph.name %>( messages, entity<%= graph.filter_parameter.nil? ? '' : ", filter" %> );
    changeSet.merge( descriptor, messages );
<%
    elsif !graph.external_data_load? && !graph.filtered?
-%>
    final org.realityforge.replicant.server.EntityMessageSet messages = new org.realityforge.replicant.server.EntityMessageSet();
    // Add to a list so that when routing we limit the number of new entities that are loaded during routing
    final java.util.LinkedList<Object> objects = new java.util.LinkedList<>();
<% graph.type_roots.each do |type_root_name|
  type = repository.entity_by_name(type_root_name)
-%>
    objects.addAll( _<%= Reality::Naming.camelize("#{type.data_module.name}#{type.name}") %>Repository.findAll() );
<% end -%>
    _encoder.encodeObjects( messages, objects );
    changeSet.merge( descriptor, messages );
<%
    else
-%>
    _context.collect<%= graph.name %>( descriptor, changeSet<%= graph.filter_parameter.nil? ? '' : ', filter' %> );
<%
    end
-%>
<% if graph.cacheable? -%>
    return <% if graph.external_cache_management? -%>_context.get<%= graph.name %>CacheKey()<% else %>java.lang.String.valueOf( java.lang.System.currentTimeMillis() )<% end %>;
<% end -%>
  }

  @java.lang.Override
  public void unsubscribeFrom<%= graph.name %>( @javax.annotation.Nonnull final String sessionID<%= param %> )
    throws <%= repository.exception_by_name(repository.imit.invalid_session_exception).ee.qualified_name %>
  {
    unsubscribe( sessionID, new org.realityforge.replicant.server.ChannelDescriptor( <%= repository.imit.qualified_graph_enum_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>.ordinal(), <%= graph.instance_root? ? pk_accessor : 'null' %> ) );
  }

<% end -%>

  @java.lang.Override
  protected java.lang.String collectDataForSubscribe( @javax.annotation.Nullable final org.realityforge.replicant.server.transport.ReplicantSession session, @javax.annotation.Nonnull final org.realityforge.replicant.server.ChannelDescriptor descriptor,@javax.annotation.Nonnull final org.realityforge.replicant.server.ChangeSet changeSet,@javax.annotation.Nullable final Object filter )
  {
    <%
  repository.imit.graphs.each do |graph|
-%>if ( <%= repository.imit.qualified_graph_enum_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>.ordinal() == descriptor.getChannelID() )
    {
      <% if graph.cacheable? -%>return <% end -%>collectDataFor<%= graph.name %>( session, descriptor, changeSet<% if graph.filter_parameter -%>, (<%= Domgen::Java.java_type(graph.filter_parameter, :ee, :default) %>) filter<% end %> );
<% unless graph.cacheable? -%>
      return null;
<% end -%>
    }
    else <% end -%>

    {
      throw new IllegalStateException( "collectDataForSubscribe called for unknown channel " + descriptor );
    }
  }

  @java.lang.Override
  protected boolean bulkCollectDataForSubscribe( @javax.annotation.Nullable final org.realityforge.replicant.server.transport.ReplicantSession session, @javax.annotation.Nonnull final java.util.ArrayList<org.realityforge.replicant.server.ChannelDescriptor> descriptors, @javax.annotation.Nonnull final org.realityforge.replicant.server.ChangeSet changeSet, @javax.annotation.Nullable final Object filter, final boolean explicitSubscribe )
  {
    if( descriptors.isEmpty() )
    {
      return true;
    }
    final org.realityforge.replicant.server.ChannelDescriptor descriptor = descriptors.get( 0 );
    <%
  repository.imit.graphs.select{|graph| graph.instance_root?}.each do |graph|
-%>if ( <%= repository.imit.qualified_graph_enum_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>.ordinal() == descriptor.getChannelID() )
    {
      return _context.bulkCollectDataFor<%= graph.name %>( session, descriptors, changeSet<% if graph.filter_parameter -%>, (<%= Domgen::Java.java_type(graph.filter_parameter, :ee, :default) %>) filter<% end %>, explicitSubscribe );
    }
    else <% end -%>

    {
      throw new IllegalStateException( "bulkCollectDataForSubscribe called for unknown channel " + descriptor );
    }
  }

  @java.lang.Override
  protected boolean bulkCollectDataForSubscriptionUpdate( @javax.annotation.Nullable final org.realityforge.replicant.server.transport.ReplicantSession session, @javax.annotation.Nonnull final java.util.ArrayList<org.realityforge.replicant.server.ChannelDescriptor> descriptors, @javax.annotation.Nonnull final org.realityforge.replicant.server.ChangeSet changeSet, @javax.annotation.Nullable final Object originalFilter, @javax.annotation.Nullable final Object filter )
  {
    if( descriptors.isEmpty() )
    {
      return true;
    }
    final org.realityforge.replicant.server.ChannelDescriptor descriptor = descriptors.get( 0 );
    <%
  repository.imit.graphs.select{|graph| graph.instance_root? && graph.filtered? }.each do |graph|
-%>if ( <%= repository.imit.qualified_graph_enum_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>.ordinal() == descriptor.getChannelID() )
    {
      return _context.bulkCollectDataFor<%= graph.name %>Update( session, descriptors, changeSet, (<%= Domgen::Java.java_type(graph.filter_parameter, :ee, :default) %>) originalFilter, (<%= Domgen::Java.java_type(graph.filter_parameter, :ee, :default) %>) filter );
    }
    else <% end -%>

    {
      throw new IllegalStateException( "bulkCollectDataForSubscriptionUpdate called for unknown channel " + descriptor );
    }
  }

  @java.lang.Override
  protected void collectDataForSubscriptionUpdate( @javax.annotation.Nonnull final org.realityforge.replicant.server.transport.ReplicantSession session,
                                                   @javax.annotation.Nonnull final org.realityforge.replicant.server.ChannelDescriptor descriptor,
                                                   @javax.annotation.Nonnull final org.realityforge.replicant.server.ChangeSet changeSet,
                                                   @javax.annotation.Nullable final Object originalFilter,
                                                   @javax.annotation.Nullable final Object filter )
  {
    <%
  repository.imit.graphs.select{|graph| graph.filter_parameter && !graph.filter_parameter.immutable?}.each do |graph|
-%>if ( <%= repository.imit.qualified_graph_enum_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>.ordinal() == descriptor.getChannelID() )
    {
      collectDataFor<%= graph.name %>SubscriptionUpdate( session, descriptor, changeSet, (<%= Domgen::Java.java_type(graph.filter_parameter, :ee, :default) %>) originalFilter, (<%= Domgen::Java.java_type(graph.filter_parameter, :ee, :default) %>) filter );
    }
    else <% end -%>

    {
      throw new IllegalStateException( "collectDataForSubscriptionUpdate called for unknown channel " + descriptor );
    }
  }

  @java.lang.Override
  protected boolean shouldFollowLink( @javax.annotation.Nonnull final org.realityforge.replicant.server.transport.SubscriptionEntry sourceEntry, @javax.annotation.Nonnull final org.realityforge.replicant.server.ChannelDescriptor target )
  {
    final org.realityforge.replicant.server.ChannelDescriptor source = sourceEntry.getDescriptor();
    final int sourceChannelID = source.getChannelID();
    final int targetChannelID = target.getChannelID();
    <% repository.imit.graphs.select{|g| g.instance_root? }.collect{|g| g.inward_graph_links.select{|graph_link| graph_link.auto?} }.flatten.each do |graph_link|
         graph = repository.imit.graph_by_name(graph_link.target_graph)
         source_graph = repository.imit.graph_by_name(graph_link.source_graph)
         entity = repository.entity_by_name(graph.instance_root)
         if graph.filter_parameter && graph.instance_root? && source_graph.instance_root?
%>if ( <%= repository.imit.qualified_graph_enum_name %>.<%= Reality::Naming.uppercase_constantize(source_graph.name) %>.ordinal() == source.getChannelID() && <%= repository.imit.qualified_graph_enum_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>.ordinal() == target.getChannelID() )
    {
      final <%= entity.primary_key.jpa.java_type %> targetID = (<%= entity.primary_key.jpa.java_type %>) target.getSubChannelID();
      final <%= entity.jpa.qualified_name %> entity = _<%= Reality::Naming.camelize("#{entity.data_module.name}#{entity.name}") %>Repository.getByID( targetID );
      return _context.shouldFollowLinkFrom<%= graph_link.source_graph %>To<%= graph.name %>( entity, (<%= Domgen::Java.java_type(graph.filter_parameter, :ee, :default) %>) sourceEntry.getFilter() );
    }
    else <%
    end
    end
-%>

    {
      throw new IllegalStateException( "shouldFollowLink called for link between channel " + source + " and " + target + " and the target has no filter or the link is unknown." );
    }
  }

  @javax.annotation.Nonnull
  @java.lang.Override
  protected java.lang.RuntimeException newBadSessionException( @javax.annotation.Nonnull final String sessionID )
  {
    return new <%= repository.exception_by_name(repository.imit.invalid_session_exception).ee.qualified_name %>();
  }
<% if repository.imit.poll_replicate_mode? -%>

  @java.lang.Override
  @javax.annotation.Nullable
  public String poll( @javax.annotation.Nonnull final String clientID, final int lastSequenceAcked )
    throws <%= repository.exception_by_name(repository.imit.invalid_session_exception).ee.qualified_name %>
  {
    return pollJsonData( ensureSession( clientID ), lastSequenceAcked );
  }
<% end -%>
}
<commit_msg>Suppress a warning that not relevant<commit_after>/* DO NOT EDIT: File is auto-generated */
package <%= to_package(repository.imit.qualified_session_manager_name) %>;

@javax.annotation.Generated( "Domgen" )
@java.lang.SuppressWarnings( { "unchecked", "PMD.UnusedPrivateField", "PMD.CollapsibleIfStatements", "PMD.UnusedFormalParameter" } )
@edu.umd.cs.findbugs.annotations.SuppressFBWarnings({"DLS_DEAD_LOCAL_STORE"})
<% if repository.ee.use_cdi? -%>
@javax.enterprise.context.ApplicationScoped
@javax.transaction.Transactional( javax.transaction.Transactional.TxType.REQUIRED )
@javax.enterprise.inject.Typed( { <%= repository.imit.qualified_poll_service_name %>.class, org.realityforge.replicant.server.EntityMessageEndpoint.class, org.realityforge.ssf.SessionManager.class, org.realityforge.replicant.server.transport.ReplicantSessionManager.class, <%= repository.service_by_name(repository.imit.subscription_manager).ejb.qualified_service_name %>.class } )
<% else -%>
@javax.ejb.Singleton( name = <%= repository.service_by_name(repository.imit.subscription_manager).ejb.qualified_service_name %>.NAME )
@javax.ejb.ConcurrencyManagement( javax.ejb.ConcurrencyManagementType.BEAN )
@javax.ejb.Local( { <%= repository.imit.qualified_poll_service_name %>.class, org.realityforge.replicant.server.EntityMessageEndpoint.class, org.realityforge.ssf.SessionManager.class, org.realityforge.replicant.server.transport.ReplicantSessionManager.class, <%= repository.service_by_name(repository.imit.subscription_manager).ejb.qualified_service_name %>.class } )
<% end -%>
public class <%= repository.imit.session_manager_name %>
  extends org.realityforge.replicant.server.transport.ReplicantSessionManagerImpl
  implements <%= repository.service_by_name(repository.imit.subscription_manager).ejb.qualified_service_name %><% if repository.imit.poll_replicate_mode? %>, <%= repository.imit.qualified_poll_service_name %><% end %>
{
<% repository.imit.graphs.each_with_index do |graph, i| -%>
  private static final org.realityforge.replicant.server.transport.ChannelMetaData $CHANNEL_<%= i %> = new org.realityforge.replicant.server.transport.ChannelMetaData( <%= repository.imit.qualified_graph_enum_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>.ordinal(), "<%= graph.name %>", <%= !graph.instance_root? %>, org.realityforge.replicant.server.transport.ChannelMetaData.FilterType.<%= graph.filter_parameter.nil? ? 'NONE' : graph.filter_parameter.immutable? ? 'STATIC' : 'DYNAMIC' %>, <%= graph.cacheable? %> );
<% end -%>
  private static final org.realityforge.replicant.server.transport.ChannelMetaData[] CHANNELS = new org.realityforge.replicant.server.transport.ChannelMetaData[]{<% repository.imit.graphs.each_with_index do |graph, i| -%> $CHANNEL_<%= i %>,<% end %> };
  private static final java.util.logging.Logger LOG = java.util.logging.Logger.getLogger( <%= repository.imit.session_manager_name %>.class.getName() );
  @javax.inject.Inject
  private <%= repository.imit.qualified_graph_encoder_name %> _encoder;
<%
entities = repository.imit.graphs.select{|g| g.instance_root?}.collect{|g|repository.entity_by_name(g.instance_root)}
repository.imit.graphs.select{|g| !g.instance_root?}.each do |graph|
  graph.type_roots.each do |type_root_name|
    entities << repository.entity_by_name(type_root_name)
  end
end

entities.uniq.each do |entity| -%>
  @javax.inject.Inject
  private <%= entity.dao.jpa.qualified_dao_service_name %> _<%= Reality::Naming.camelize("#{entity.data_module.name}#{entity.name}") %>Repository;
<% end -%>
  @javax.inject.Inject
  private <%= repository.service_by_name(repository.imit.session_context_service).ejb.qualified_service_name %> _context;
  @javax.annotation.Resource
  private javax.transaction.TransactionSynchronizationRegistry _registry;

  @javax.annotation.Nonnull
  @java.lang.Override
  protected javax.transaction.TransactionSynchronizationRegistry getRegistry()
  {
    return _registry;
  }

  @javax.annotation.Nonnull
  @java.lang.Override
  protected org.realityforge.replicant.server.transport.ChannelMetaData[] getChannelMetaData()
  {
    return CHANNELS;
  }

  /**
   * Remove idle session changes. This is called by a scheduler.
   */
  @java.lang.Override
  public void removeIdleSessions()
  {
    final int maxIdleTime = 1000 * 60 * 5;
    final int removedSessions = removeIdleSessions( maxIdleTime );
    if ( 0 != removedSessions && LOG.isLoggable( java.util.logging.Level.INFO ) )
    {
      LOG.info( "Removed " + removedSessions + " idle sessions" );
    }
  }

  @java.lang.Override
  protected void processDeleteMessages( @javax.annotation.Nonnull final org.realityforge.replicant.server.EntityMessage message, @javax.annotation.Nonnull final java.util.Collection<org.realityforge.replicant.server.transport.ReplicantSession> sessions )
  {
    final java.util.Map<String, java.io.Serializable> routingKeys = message.getRoutingKeys();
<%
  repository.imit.graphs.select{|graph| graph.instance_root?}.each do |graph|

    instance_root = repository.entity_by_name(graph.instance_root)
    type = instance_root.primary_key.jpa.non_primitive_java_type
    %>
    final <%= type %> <%= graph.name %> = (<%= type %>) routingKeys.get( <%= repository.imit.qualified_router_impl_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>_KEY );
    if ( null != <%= graph.name %> )
    {
      final org.realityforge.replicant.server.ChannelDescriptor descriptor = new org.realityforge.replicant.server.ChannelDescriptor( <%= repository.imit.qualified_graph_enum_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>.ordinal(), <%= graph.instance_root? ? graph.name : 'null' %> );
<%
  extra_interest_filter = ''
  if graph.filtered?
    graph.routing_keys.each do |routing_key|
      var_name = Reality::Naming.camelize(routing_key.name)
      extra_interest_filter << ", #{var_name}"
      param_type = routing_key.target_attribute.jpa.non_primitive_java_type
      param_type = "java.util.List<#{param_type}>" if routing_key.multivalued?
-%>
      final <%= param_type %> <%= var_name %> = (<%= param_type %>) routingKeys.get( <%= repository.imit.qualified_router_impl_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>_<%= Reality::Naming.uppercase_constantize(routing_key.name) %>_KEY );
<%
    end
  end

  customize_filter = ", session#{graph.instance_root? ? ", #{graph.name}" : ''}"
  customize_filter = "#{customize_filter}, (#{Domgen::Java.java_type(graph.filter_parameter, :ee, :default)}) entry.getFilter()" if graph.filter_parameter?
  customize_filter = "#{customize_filter}#{extra_interest_filter}"
-%>
      for ( final org.realityforge.replicant.server.transport.ReplicantSession session : sessions )
      {
        final org.realityforge.replicant.server.transport.SubscriptionEntry entry = session.findSubscriptionEntry( descriptor );
        if ( null != entry )
        {
          final org.realityforge.replicant.server.EntityMessage m = _context.filterMessageOfInterestIn<%= graph.name %>Graph( message<%= customize_filter %> );
          if( null != m )
          {
            if( m.isDelete() && <%= repository.imit.qualified_message_constants_name %>.<%= Reality::Naming.uppercase_constantize(instance_root.data_module.name.to_s) %>_<%= Reality::Naming.uppercase_constantize(instance_root.name.to_s) %> == m.getTypeID() )
            {
              performUnsubscribe( session, entry, true );
            }
          }
        }
      }
    }
<% end -%>
  }

  @java.lang.Override
  protected void processUpdateMessages( @javax.annotation.Nonnull final org.realityforge.replicant.server.EntityMessage message, @javax.annotation.Nonnull final java.util.Collection<org.realityforge.replicant.server.transport.ReplicantSession> sessions, @javax.annotation.Nonnull org.realityforge.replicant.server.ChangeAccumulator accumulator )
  {
    final java.util.Map<String, java.io.Serializable> routingKeys = message.getRoutingKeys();
<%
  repository.imit.graphs.each do |graph|

  type = 'Boolean'
  if graph.instance_root?
    instance_root = repository.entity_by_name(graph.instance_root)
    type = instance_root.primary_key.jpa.non_primitive_java_type
  end
-%>
    final <%= type %> <%= graph.name %> = (<%= type %>) routingKeys.get( <%= repository.imit.qualified_router_impl_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>_KEY );
    if ( null != <%= graph.name %> )
    {
      final org.realityforge.replicant.server.ChannelDescriptor descriptor = new org.realityforge.replicant.server.ChannelDescriptor( <%= repository.imit.qualified_graph_enum_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>.ordinal(), <%= graph.instance_root? ? graph.name : 'null' %> );
<%
  extra_interest_filter = ''
  if graph.filtered?
    graph.routing_keys.each do |routing_key|
      var_name = Reality::Naming.camelize(routing_key.name)
      extra_interest_filter << ", #{var_name}"
      param_type = routing_key.target_attribute.jpa.non_primitive_java_type
      param_type = "java.util.List<#{param_type}>" if routing_key.multivalued?
-%>
      final <%= param_type %> <%= var_name %> = (<%= param_type %>) routingKeys.get( <%= repository.imit.qualified_router_impl_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>_<%= Reality::Naming.uppercase_constantize(routing_key.name) %>_KEY );
<%
    end
  end

  customize_filter = ", session#{graph.instance_root? ? ", #{graph.name}" : ''}"
  customize_filter = "#{customize_filter}, (#{Domgen::Java.java_type(graph.filter_parameter, :ee, :default)}) entry.getFilter()" if graph.filter_parameter?
  customize_filter = "#{customize_filter}#{extra_interest_filter}"
-%>
<% if graph.cacheable? && !graph.external_cache_management? -%>
      deleteCacheEntry( descriptor );
<% end -%>
      for ( final org.realityforge.replicant.server.transport.ReplicantSession session : sessions )
      {
        final org.realityforge.replicant.server.transport.SubscriptionEntry entry = session.findSubscriptionEntry( descriptor );
        if ( null != entry )
        {
          final org.realityforge.replicant.server.EntityMessage m = _context.filterMessageOfInterestIn<%= graph.name %>Graph( message<%= customize_filter %> );
          if( null != m )
          {
            accumulator.addChange( session, new org.realityforge.replicant.server.Change( m, descriptor.getChannelID(), descriptor.getSubChannelID() ) );
          }
        }
      }
    }
<% end -%>
  }

<%
  repository.imit.graphs.each do |graph|

  param = []
  param_value = []
  pk_accessor = ''
  if graph.instance_root?
    instance_root = repository.entity_by_name(graph.instance_root)
    param << "@javax.annotation.Nonnull final #{instance_root.jpa.qualified_name} entity"
    param_value << 'entity'
    pk_accessor = "entity.#{getter_for(instance_root.primary_key)}"
  end
  update_param = param.dup
  update_param_value = param_value.dup
  filter_param = nil
  if graph.filter_parameter
    p = graph.filter_parameter
    filter_param = "#{nullability_annotation(p.nullable?)} #{Domgen::Java.java_type(p, :ee, :default)} filter"
    update_param << filter_param
    update_param_value << 'filter'
  end
  param = param.join(', ')
  update_param = update_param.join(', ')

  param = ", #{param}" if param.size > 0
  update_param = ", #{update_param}" if update_param.size > 0
-%>

  @java.lang.Override
  public <%= graph.cacheable? ? 'boolean': 'void' %> subscribeTo<%= graph.name %>( @javax.annotation.Nonnull final String sessionID<%= update_param %><% if graph.cacheable? -%>, @javax.annotation.Nullable final String cacheKey<% end %> )
    throws <%= repository.exception_by_name(repository.imit.invalid_session_exception).ee.qualified_name %>
  {
    <%= graph.cacheable? ? 'final CacheStatus status = ': '' %>subscribe( sessionID, new org.realityforge.replicant.server.ChannelDescriptor( <%= repository.imit.qualified_graph_enum_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>.ordinal(), <%= graph.instance_root? ? "#{pk_accessor}" : 'null' %> ), <%= graph.filter_parameter.nil? ? 'null' : 'filter' %>, <%= graph.cacheable? ? 'cacheKey' : 'null' %> );
<% if graph.cacheable? -%>
    return status != CacheStatus.USE;
<% end -%>
  }
<% if graph.instance_root?

  root_type = Domgen::Java.java_type(instance_root.primary_key, :ee, :default)
-%>

  public void subscribeToMultiple<%= Reality::Naming.pluralize(graph.name) %>( @javax.annotation.Nonnull final String sessionID, @javax.annotation.Nonnull final java.util.List<<%= root_type %>> roots<%= filter_param.nil? ? '' : ", #{filter_param}" %> )
    throws <%= repository.exception_by_name(repository.imit.invalid_session_exception).ee.qualified_name %>
  {
    bulkSubscribe( sessionID, <%= repository.imit.qualified_graph_enum_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>.ordinal(), (java.util.Collection<java.io.Serializable>) (java.util.Collection) roots, <%= filter_param.nil? ? 'null' : 'filter' %>, true );
  }

  public void unsubscribeFromMultiple<%= Reality::Naming.pluralize(graph.name) %>( @javax.annotation.Nonnull final String sessionID, @javax.annotation.Nonnull final java.util.List<<%= root_type %>> roots )
    throws <%= repository.exception_by_name(repository.imit.invalid_session_exception).ee.qualified_name %>
  {
    bulkUnsubscribe( sessionID, <%= repository.imit.qualified_graph_enum_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>.ordinal(), (java.util.Collection<java.io.Serializable>) (java.util.Collection) roots, true );
  }
<% end -%>
<% if !graph.instance_root? && !graph.external_data_load? -%>
  private void collect<%= graph.name %>( @javax.annotation.Nonnull org.realityforge.replicant.server.EntityMessageSet messages )
  {
    // Add to a list so that when routing we limit the number of new entities that are loaded during routing
    final java.util.LinkedList<Object> objects = new java.util.LinkedList<>();
<% graph.type_roots.each do |type_root_name|
  type = repository.entity_by_name(type_root_name)
-%>
    objects.addAll( _<%= Reality::Naming.camelize("#{type.data_module.name}#{type.name}") %>Repository.findAll() );
<% end -%>
    _encoder.encodeObjects( messages, objects );
  }

<% end -%>
<% if graph.instance_root? -%>

  private org.realityforge.replicant.server.EntityMessageSet encode<%= graph.name %>( @javax.annotation.Nonnull final <%= repository.entity_by_name(graph.instance_root).jpa.qualified_name %> entity<%= graph.filter_parameter.nil? ? '' : ", #{Domgen::Java.java_type(graph.filter_parameter, :ee, :default)} filter" %> )
  {
    final org.realityforge.replicant.server.EntityMessageSet messages = new org.realityforge.replicant.server.EntityMessageSet();
    _encoder.encode<%= graph.name %>( messages, entity<%= graph.filter_parameter.nil? ? '' : ", filter" %> );
    return messages;
  }
<% end -%>
<% if graph.filter_parameter && !graph.filter_parameter.immutable? -%>
<% if graph.instance_root? -%>
  @java.lang.Override
  public void updateMultiple<%= graph.name %>Subscriptions( @javax.annotation.Nonnull final String sessionID, @javax.annotation.Nonnull final java.util.List<<%= root_type %>> roots, <%= filter_param %> )
    throws <%= repository.exception_by_name(repository.imit.invalid_session_exception).ee.qualified_name %>
  {
    bulkUpdateSubscription( sessionID, <%= repository.imit.qualified_graph_enum_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>.ordinal(), (java.util.Collection<java.io.Serializable>) (java.util.Collection) roots, filter );
<%
  processed = []
  graph.outward_graph_links.select{|graph_link| graph_link.auto?}.each do |graph_link|
     target_graph = repository.imit.graph_by_name(graph_link.target_graph)
     next unless target_graph.filtered?
     key = "#{graph_link.source_graph}=>#{graph_link.target_graph}"
     next if processed.include?(key)
     processed << key
     instance_root = repository.entity_by_name(target_graph.instance_root)
     entity = repository.entity_by_name(graph.instance_root)
-%>
    for ( final <%= root_type %> id : roots )
    {
      final <%= entity.jpa.qualified_name %> entity = _<%= Reality::Naming.camelize("#{entity.data_module.name}#{entity.name}") %>Repository.getBy<%= entity.primary_key.name %>( id );
      //Propogate the filter from <%= graph.name %> to downstream graph <%= target_graph.name %>
      for ( final <%= instance_root.jpa.qualified_name %> downstreamEntity : _context.getLinksToUpdateFor<%= graph_link.source_graph %>To<%= target_graph.name %>( entity, filter ) )
      {
        subscribeTo<%= target_graph.name %>( sessionID, downstreamEntity, filter );
      }
    }
<% end -%>
  }
<% end -%>

  @java.lang.Override
  public void update<%= graph.name %>Subscription( @javax.annotation.Nonnull final String sessionID<%= update_param %> )
    throws <%= repository.exception_by_name(repository.imit.invalid_session_exception).ee.qualified_name %>
  {
    updateSubscription( sessionID, new org.realityforge.replicant.server.ChannelDescriptor( <%= repository.imit.qualified_graph_enum_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>.ordinal(), <%= graph.instance_root? ? pk_accessor : 'null' %> ), filter );
<%
  processed = []
  graph.outward_graph_links.select{|graph_link| graph_link.auto?}.each do |graph_link|
     target_graph = repository.imit.graph_by_name(graph_link.target_graph)
     next unless target_graph.filtered?
     key = "#{graph_link.source_graph}=>#{graph_link.target_graph}"
      next if processed.include?(key)
      processed << key
      instance_root = repository.entity_by_name(target_graph.instance_root)
-%>
    //Propogate the filter from <%= graph.name %> to downstream graph <%= target_graph.name %>
    for ( final <%= instance_root.jpa.qualified_name %> downstreamEntity : _context.getLinksToUpdateFor<%= graph_link.source_graph %>To<%= target_graph.name %>( entity, filter ) )
    {
      subscribeTo<%= target_graph.name %>( sessionID, downstreamEntity, filter );
    }
<% end if graph.instance_root? -%>
  }

  protected void collectDataFor<%= graph.name %>SubscriptionUpdate( @javax.annotation.Nonnull final org.realityforge.replicant.server.transport.ReplicantSession session, @javax.annotation.Nonnull final org.realityforge.replicant.server.ChannelDescriptor descriptor, @javax.annotation.Nonnull final org.realityforge.replicant.server.ChangeSet changeSet, final <%= Domgen::Java.java_type(graph.filter_parameter, :ee, :default) %> originalFilter, final <%= Domgen::Java.java_type(graph.filter_parameter, :ee, :default) %> filter )
  {
    final org.realityforge.replicant.server.EntityMessageSet messages = new org.realityforge.replicant.server.EntityMessageSet();
<%
    if graph.instance_root?
      entity = repository.entity_by_name(graph.instance_root)
-%>
    final <%= entity.primary_key.jpa.java_type %> id = (<%= entity.primary_key.jpa.java_type %>) descriptor.getSubChannelID();
    final <%= entity.jpa.qualified_name %> entity = _<%= Reality::Naming.camelize("#{entity.data_module.name}#{entity.name}") %>Repository.getBy<%= entity.primary_key.name %>( id );
<% end %>
    _context.collectForFilterChange<%= graph.name %>( messages, <%= graph.instance_root? ? 'entity, ' : '' %>originalFilter, filter );
    changeSet.mergeAll( org.realityforge.replicant.server.ChangeUtil.toChanges( messages.getEntityMessages(), descriptor.getChannelID(), descriptor.getSubChannelID() ) );
  }
<% end -%>

  protected <% if graph.cacheable? -%>java.lang.String<% else -%>void<% end -%> collectDataFor<%= graph.name %>( @javax.annotation.<% if graph.cacheable? -%>Nullable<% else %>Nonnull<% end %> final org.realityforge.replicant.server.transport.ReplicantSession session, @javax.annotation.Nonnull final org.realityforge.replicant.server.ChannelDescriptor descriptor, @javax.annotation.Nonnull final org.realityforge.replicant.server.ChangeSet changeSet<% if graph.filter_parameter -%>, final <%= Domgen::Java.java_type(graph.filter_parameter, :ee, :default) %> filter<% end %>)
  {
<%
    if graph.instance_root?
      entity = repository.entity_by_name(graph.instance_root)
-%>
    final org.realityforge.replicant.server.EntityMessageSet messages = new org.realityforge.replicant.server.EntityMessageSet();
    final <%= entity.primary_key.jpa.java_type %> id = (<%= entity.primary_key.jpa.java_type %>) descriptor.getSubChannelID();
    final <%= entity.jpa.qualified_name %> entity = _<%= Reality::Naming.camelize("#{entity.data_module.name}#{entity.name}") %>Repository.getBy<%= entity.primary_key.name %>( id );
    _encoder.encode<%= graph.name %>( messages, entity<%= graph.filter_parameter.nil? ? '' : ", filter" %> );
    changeSet.merge( descriptor, messages );
<%
    elsif !graph.external_data_load? && !graph.filtered?
-%>
    final org.realityforge.replicant.server.EntityMessageSet messages = new org.realityforge.replicant.server.EntityMessageSet();
    // Add to a list so that when routing we limit the number of new entities that are loaded during routing
    final java.util.LinkedList<Object> objects = new java.util.LinkedList<>();
<% graph.type_roots.each do |type_root_name|
  type = repository.entity_by_name(type_root_name)
-%>
    objects.addAll( _<%= Reality::Naming.camelize("#{type.data_module.name}#{type.name}") %>Repository.findAll() );
<% end -%>
    _encoder.encodeObjects( messages, objects );
    changeSet.merge( descriptor, messages );
<%
    else
-%>
    _context.collect<%= graph.name %>( descriptor, changeSet<%= graph.filter_parameter.nil? ? '' : ', filter' %> );
<%
    end
-%>
<% if graph.cacheable? -%>
    return <% if graph.external_cache_management? -%>_context.get<%= graph.name %>CacheKey()<% else %>java.lang.String.valueOf( java.lang.System.currentTimeMillis() )<% end %>;
<% end -%>
  }

  @java.lang.Override
  public void unsubscribeFrom<%= graph.name %>( @javax.annotation.Nonnull final String sessionID<%= param %> )
    throws <%= repository.exception_by_name(repository.imit.invalid_session_exception).ee.qualified_name %>
  {
    unsubscribe( sessionID, new org.realityforge.replicant.server.ChannelDescriptor( <%= repository.imit.qualified_graph_enum_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>.ordinal(), <%= graph.instance_root? ? pk_accessor : 'null' %> ) );
  }

<% end -%>

  @java.lang.Override
  protected java.lang.String collectDataForSubscribe( @javax.annotation.Nullable final org.realityforge.replicant.server.transport.ReplicantSession session, @javax.annotation.Nonnull final org.realityforge.replicant.server.ChannelDescriptor descriptor,@javax.annotation.Nonnull final org.realityforge.replicant.server.ChangeSet changeSet,@javax.annotation.Nullable final Object filter )
  {
    <%
  repository.imit.graphs.each do |graph|
-%>if ( <%= repository.imit.qualified_graph_enum_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>.ordinal() == descriptor.getChannelID() )
    {
      <% if graph.cacheable? -%>return <% end -%>collectDataFor<%= graph.name %>( session, descriptor, changeSet<% if graph.filter_parameter -%>, (<%= Domgen::Java.java_type(graph.filter_parameter, :ee, :default) %>) filter<% end %> );
<% unless graph.cacheable? -%>
      return null;
<% end -%>
    }
    else <% end -%>

    {
      throw new IllegalStateException( "collectDataForSubscribe called for unknown channel " + descriptor );
    }
  }

  @java.lang.Override
  protected boolean bulkCollectDataForSubscribe( @javax.annotation.Nullable final org.realityforge.replicant.server.transport.ReplicantSession session, @javax.annotation.Nonnull final java.util.ArrayList<org.realityforge.replicant.server.ChannelDescriptor> descriptors, @javax.annotation.Nonnull final org.realityforge.replicant.server.ChangeSet changeSet, @javax.annotation.Nullable final Object filter, final boolean explicitSubscribe )
  {
    if( descriptors.isEmpty() )
    {
      return true;
    }
    final org.realityforge.replicant.server.ChannelDescriptor descriptor = descriptors.get( 0 );
    <%
  repository.imit.graphs.select{|graph| graph.instance_root?}.each do |graph|
-%>if ( <%= repository.imit.qualified_graph_enum_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>.ordinal() == descriptor.getChannelID() )
    {
      return _context.bulkCollectDataFor<%= graph.name %>( session, descriptors, changeSet<% if graph.filter_parameter -%>, (<%= Domgen::Java.java_type(graph.filter_parameter, :ee, :default) %>) filter<% end %>, explicitSubscribe );
    }
    else <% end -%>

    {
      throw new IllegalStateException( "bulkCollectDataForSubscribe called for unknown channel " + descriptor );
    }
  }

  @java.lang.Override
  protected boolean bulkCollectDataForSubscriptionUpdate( @javax.annotation.Nullable final org.realityforge.replicant.server.transport.ReplicantSession session, @javax.annotation.Nonnull final java.util.ArrayList<org.realityforge.replicant.server.ChannelDescriptor> descriptors, @javax.annotation.Nonnull final org.realityforge.replicant.server.ChangeSet changeSet, @javax.annotation.Nullable final Object originalFilter, @javax.annotation.Nullable final Object filter )
  {
    if( descriptors.isEmpty() )
    {
      return true;
    }
    final org.realityforge.replicant.server.ChannelDescriptor descriptor = descriptors.get( 0 );
    <%
  repository.imit.graphs.select{|graph| graph.instance_root? && graph.filtered? }.each do |graph|
-%>if ( <%= repository.imit.qualified_graph_enum_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>.ordinal() == descriptor.getChannelID() )
    {
      return _context.bulkCollectDataFor<%= graph.name %>Update( session, descriptors, changeSet, (<%= Domgen::Java.java_type(graph.filter_parameter, :ee, :default) %>) originalFilter, (<%= Domgen::Java.java_type(graph.filter_parameter, :ee, :default) %>) filter );
    }
    else <% end -%>

    {
      throw new IllegalStateException( "bulkCollectDataForSubscriptionUpdate called for unknown channel " + descriptor );
    }
  }

  @java.lang.Override
  protected void collectDataForSubscriptionUpdate( @javax.annotation.Nonnull final org.realityforge.replicant.server.transport.ReplicantSession session,
                                                   @javax.annotation.Nonnull final org.realityforge.replicant.server.ChannelDescriptor descriptor,
                                                   @javax.annotation.Nonnull final org.realityforge.replicant.server.ChangeSet changeSet,
                                                   @javax.annotation.Nullable final Object originalFilter,
                                                   @javax.annotation.Nullable final Object filter )
  {
    <%
  repository.imit.graphs.select{|graph| graph.filter_parameter && !graph.filter_parameter.immutable?}.each do |graph|
-%>if ( <%= repository.imit.qualified_graph_enum_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>.ordinal() == descriptor.getChannelID() )
    {
      collectDataFor<%= graph.name %>SubscriptionUpdate( session, descriptor, changeSet, (<%= Domgen::Java.java_type(graph.filter_parameter, :ee, :default) %>) originalFilter, (<%= Domgen::Java.java_type(graph.filter_parameter, :ee, :default) %>) filter );
    }
    else <% end -%>

    {
      throw new IllegalStateException( "collectDataForSubscriptionUpdate called for unknown channel " + descriptor );
    }
  }

  @java.lang.Override
  protected boolean shouldFollowLink( @javax.annotation.Nonnull final org.realityforge.replicant.server.transport.SubscriptionEntry sourceEntry, @javax.annotation.Nonnull final org.realityforge.replicant.server.ChannelDescriptor target )
  {
    final org.realityforge.replicant.server.ChannelDescriptor source = sourceEntry.getDescriptor();
    final int sourceChannelID = source.getChannelID();
    final int targetChannelID = target.getChannelID();
    <% repository.imit.graphs.select{|g| g.instance_root? }.collect{|g| g.inward_graph_links.select{|graph_link| graph_link.auto?} }.flatten.each do |graph_link|
         graph = repository.imit.graph_by_name(graph_link.target_graph)
         source_graph = repository.imit.graph_by_name(graph_link.source_graph)
         entity = repository.entity_by_name(graph.instance_root)
         if graph.filter_parameter && graph.instance_root? && source_graph.instance_root?
%>if ( <%= repository.imit.qualified_graph_enum_name %>.<%= Reality::Naming.uppercase_constantize(source_graph.name) %>.ordinal() == source.getChannelID() && <%= repository.imit.qualified_graph_enum_name %>.<%= Reality::Naming.uppercase_constantize(graph.name) %>.ordinal() == target.getChannelID() )
    {
      final <%= entity.primary_key.jpa.java_type %> targetID = (<%= entity.primary_key.jpa.java_type %>) target.getSubChannelID();
      final <%= entity.jpa.qualified_name %> entity = _<%= Reality::Naming.camelize("#{entity.data_module.name}#{entity.name}") %>Repository.getByID( targetID );
      return _context.shouldFollowLinkFrom<%= graph_link.source_graph %>To<%= graph.name %>( entity, (<%= Domgen::Java.java_type(graph.filter_parameter, :ee, :default) %>) sourceEntry.getFilter() );
    }
    else <%
    end
    end
-%>

    {
      throw new IllegalStateException( "shouldFollowLink called for link between channel " + source + " and " + target + " and the target has no filter or the link is unknown." );
    }
  }

  @javax.annotation.Nonnull
  @java.lang.Override
  protected java.lang.RuntimeException newBadSessionException( @javax.annotation.Nonnull final String sessionID )
  {
    return new <%= repository.exception_by_name(repository.imit.invalid_session_exception).ee.qualified_name %>();
  }
<% if repository.imit.poll_replicate_mode? -%>

  @java.lang.Override
  @javax.annotation.Nullable
  public String poll( @javax.annotation.Nonnull final String clientID, final int lastSequenceAcked )
    throws <%= repository.exception_by_name(repository.imit.invalid_session_exception).ee.qualified_name %>
  {
    return pollJsonData( ensureSession( clientID ), lastSequenceAcked );
  }
<% end -%>
}

{'fraction_non_alphanumeric': 0.14441203433619468, 'fraction_numerical': 0.0004000333361113426, 'mean_word_length': 4.791119691119691, 'pattern_counts': {'":': 0, '<': 595, '<?xml version=': 0, '>': 567, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1886672', 'n_tokens_mistral': 19446, 'n_tokens_neox': 18308, 'n_words': 2896}