This last recipe is an extension of the Spring Security module that is applied to asynchronous services and controllers. Follow these steps on how threads in asynchronous and reactive executions can access the user details at runtime:

1. Before this recipe starts, include inside the pom.xml all the needed Maven dependencies of Spring Security 4.2.2. Refer to Chapter 4, Securing Spring MVC Applications, recipe ;Applying Aspect-Oriented Programming, for this item.
2. Create a new package org.packt.web.reactor.security.config to contain the Security context definition derived from Chapter 4, Securing Spring MVC Applications, recipe Applying Aspect-Oriented Programming. The inMemoryAuthentication() security configuration will be taken as the security protocol for this recipe. Also, there will be a slight modification on the http.sessionManagement() operation to consider asynchronous request executions:

@Configuration 
@EnableWebSecurity 
public class AppSecurityConfig extends  
      WebSecurityConfigurerAdapter { 
      @Override 
    protected void configure(AuthenticationManagerBuilder auth)  
         throws Exception { 
       auth.inMemoryAuthentication() 
          .withUser("sjctrags") 
.password("sjctrags").roles("USER"); 
    } 
    
    @Override 
    protected void configure(HttpSecurity http) throws Exception { 
        http 
          .authorizeRequests() 
          .antMatchers("/react/login**",  
                     "/react/after**").permitAll() 
          .anyRequest().authenticated() 
          .and() 
          .formLogin() 
          .loginPage("/react/login.html") 
          .defaultSuccessUrl("/react/menu.html") 
          .failureUrl("/react/login.html?error=true") 
          .and().logout().logoutUrl("/react/logout.html") 
          .logoutSuccessUrl("/react/after_logout.html") 
          .and().sessionManagement() 
          .sessionCreationPolicy( 
SessionCreationPolicy.IF_REQUIRED); 
           
         http.csrf().disable();     
    } 
    
    @Override 
    public void configure(WebSecurity web) throws Exception { 
      web 
        .ignoring() 
           .antMatchers("/resources/**") 
           .antMatchers("/css/**") 
           .antMatchers("/js/**") 
           .antMatchers("/image/**"); 
    } 
} 

3. Also include in the new package the configuration of Spring Security's DelegatingFilterProxy with some added support on asynchronous and non-blocking transactions:

public class SpringSecurityInitializer extends  
      AbstractSecurityWebApplicationInitializer { 
    
   @Override 
   protected boolean isAsyncSecuritySupported() { 
      return true; 
   } 
    
   @Override 
   protected EnumSet<DispatcherType>  
getSecurityDispatcherTypes() { 
      return EnumSet.of(DispatcherType.ASYNC,  
         DispatcherType.REQUEST, DispatcherType.FORWARD,  
         DispatcherType.INCLUDE); 
   } 
} 

4. Create a service class EmployeeParallelStreamService inside the same package as the previous service classes. Add the following version of showAllEmployees() that uses parallelStream() to forEach() all employee records:

public void showAllEmployees(){ 
Consumer<Employee> showAll =  
(e) -> { 
System.out.format("%s %s %d
",  
e.getFirstName(), e.getLastName(), e.getAge()); 
}; 
employeeDaoImpl.getEmployees() 
.parallelStream() 
.forEach(showAll); 
} 

5. Now, apply the security protocol by integrating the AppSecurityConfig context definition to SpringContextConfig through the @Import annotation:

@Import(value = { AppSecurityConfig.class }) 
@Configuration 
@EnableWebMvc 
@ComponentScan(basePackages = {"org.packt.web.reactor",  
      "org.packt.web.reactor.model"}) 
public class SpringContextConfig  { } 

6. For the thread pool to access SecurityContext for further authentication and authorization rules, inject the following org.springframework.beans.factory.config.MethodInvokingFactoryBean with the necessary details to SpringContextConfig definition. The SecurityContext is only accessible by the main thread and not with the thread pool generated by TaskExecutors:

@Bean 
   public MethodInvokingFactoryBean  
         methodInvokingFactoryBean() { 
     MethodInvokingFactoryBean methodInvokingFactoryBean =  
new MethodInvokingFactoryBean(); 
     methodInvokingFactoryBean 
.setTargetClass(SecurityContextHolder.class); 
     methodInvokingFactoryBean 
.setTargetMethod("setStrategyName"); 
     methodInvokingFactoryBean 
.setArguments(new String[]{ 
SecurityContextHolder.MODE_INHERITABLETHREADLOCAL}); 
     return methodInvokingFactoryBean; 
}

7. Open EmployeeServiceImpl and DepartmentServiceImpl and apply some audit logs to check if the java.security.Principal is propagated by all asynchronous threads. First, implement readEmployees() method that processes employee record retrieval for 6000 milliseconds and returns CompletableFuture<T>:

@Service 
public class EmployeeServiceImpl implements EmployeeService { 
    
   @Async 
   @Override 
   public CompletableFuture<List<Employee>> readEmployees() { 
       
        Supplier<List<Employee>> supplyListEmp = ()->{ 
         // refer to sources 
         try { 
            System.out.println("readEmployees Callable  
            login: " + SecurityContextHolder.getContext() 
.getAuthentication().getPrincipal()); 
            Thread.sleep(6000); 
         } catch (InterruptedException e) { } 
         return employeeDaoImpl.getEmployees(); 
      }; 
      return CompletableFuture.supplyAsync(supplyListEmp); 
   } 
} 

8. Implement the readEmployeesCall() that acceses the user credentials and returns the Callable<t> task:

    
@Override 
public Callable<List<Employee>> readEmployeesCall() { 
  Callable< List<Employee> > task = new Callable< List<Employee> >() { 
 
    @Override 
    public  List<Employee>  call () throws  
    Exception { 
      // refer to sources 
      System.out.println("readEmployeesCall Callable  
      login: " + SecurityContextHolder.getContext() 
      .getAuthentication().getPrincipal()); 
      Thread.sleep(6000); 
      List<Employee> empList = employeeDaoImpl.getEmployees(); 
      return empList; 
    } 
  }; 
  return task; 
} 

9. Add an @Async method that accesses the user credentials and processes the insertion of Employee record within 1000 milliseconds:

@Async 
@Override 
public void addEmployee(EmployeeForm empForm) { 
  // refer to sources 
  try { 
    // refer to sources 
    System.out.println("addEmployee @Async login: " +  
    SecurityContextHolder.getContext() 
    .getAuthentication().getPrincipal()); 
    Thread.sleep(1000); 
  } catch (InterruptedException e) { } 
  employeeDaoImpl.addEmployeeBySJI(emp); 
} 

10. Methods that return Future<T> can also access the security user details:

@Async 
public Future<Employee> readEmployee(Integer empId) { 
  try { 
  System.out.println("service:readEmployee(empid)  
  task executor: " + Thread.currentThread().getName()); 
  System.out.println("processing for 2000 ms"); 
  System.out.println("readEmployee @Async login: " +  
  SecurityContextHolder.getContext() 
  .getAuthentication().getPrincipal()); 
  Thread.sleep(2000); 
  } catch (InterruptedException e) { } 
  return new AsyncResult<>(employeeDaoImpl.getEmployee(empId)); 
}

11. Lastly, methods that return the non-blocking and reactive Flux<T> and Mono<T> streams can also access user credentials:

@Override 
public Flux<Employee> readEmployeesFlux(int age) { 
  Scheduler subWorker = Schedulers.newSingle("sub-thread"); 
  Scheduler pubWorker = Schedulers.newSingle("pub-thread"); 
  Predicate<Employee> validAge = (e) -> { 
    // refer to sources 
    System.out.println("flux:filter task executor  
    login: " + SecurityContextHolder.getContext() 
    .getAuthentication().getPrincipal()); 
    return e.getAge() > age; 
  };       
  Supplier<Flux<Employee>> deferredTask = ()->{ 
    // refer to sources 
    System.out.println("flux:defer task executor  
    login: " + SecurityContextHolder.getContext() 
    .getAuthentication().getPrincipal()); 
    return Flux.fromIterable(employeeDaoImpl.getEmployees()); 
  }; 
  Flux<Employee> deferred = Flux.defer(deferredTask) 
  .filter(validAge).subscribeOn(subWorker) 
  .publishOn(pubWorker); 
  return deferred; 
} 

12. Save all files. Delete the existing ch08 context folder in Tomcat's /webapps before the new deployment. Then clean, install, and deploy the new project with Spring Security 4.2.x. Run https://localhost:8443/react/login.html and log in using the given credentials.
13. After executing the services, open Tomcat's log and observe a similar output:

readDepartments CompletableFuture login: org.springframework.security.core.userdetails.User@41036863: Username: sjctrags; Password: [PROTECTED]; Enabled: true; AccountNonExpired: true; credentialsNonExpired: true; AccountNonLocked: true; Granted Authorities: ROLE_USER
    
addDepartment @Async login: org.springframework.security.core.userdetails.User@41036863: Username: sjctrags; Password: [PROTECTED]; Enabled: true; AccountNonExpired: true; credentialsNonExpired: true; AccountNonLocked: true; Granted Authorities: ROLE_USER
    
readEmployees Callable login: org.springframework.security.core.userdetails.User@41036863: Username: sjctrags; Password: [PROTECTED]; Enabled: true; AccountNonExpired: true; credentialsNonExpired: true; AccountNonLocked: true; Granted Authorities: ROLE_USER