vlambda博客
学习文章列表

    <dependency> 
      <groupId>org.reactivestreams</groupId> 
      <artifactId>reactive-streams</artifactId> 
      <version>1.0.1</version> 
   </dependency> 
   <dependency> 
      <groupId>org.reactivestreams</groupId> 
      <artifactId>reactive-streams-tck</artifactId> 
      <version>1.0.1</version> 
   </dependency> 

    <parent> 
       <groupId>org.springframework.boot</groupId> 
       <artifactId>spring-boot-starter-parent</artifactId> 
       <version>2.0.0.M3</version> 
       <relativePath/> <!-- lookup parent from repository --> 
    </parent> 
 
    <properties> 
       <project.build.sourceEncoding>UTF-
        8</project.build.sourceEncoding> 
       <project.reporting.outputEncoding>UTF
        -8</project.reporting.outputEncoding> 
       <java.version>1.8</java.version> 
    </properties> 
 
    <dependencies> 
       <dependency> 
          <groupId>org.springframework.boot</groupId> 
          <artifactId>spring-boot-starter-webflux</artifactId> 
       </dependency> 
 
       <dependency> 
          <groupId>org.springframework.boot</groupId> 
          <artifactId>spring-boot-starter-test</artifactId> 
          <scope>test</scope> 
       </dependency> 
       <dependency> 
          <groupId>io.projectreactor</groupId> 
          <artifactId>reactor-test</artifactId> 
          <scope>test</scope> 
       </dependency> 
    </dependencies> 

    package com.packt.patterninspring.chapter11.
      reactivewebapp.controller; 
 
    import org.reactivestreams.Publisher; 
    import org.springframework.beans.factory.annotation.Autowired; 
    import org.springframework.web.bind.annotation.GetMapping; 
    import org.springframework.web.bind.annotation.PathVariable; 
    import org.springframework.web.bind.annotation.PostMapping; 
    import org.springframework.web.bind.annotation.RequestBody; 
    import org.springframework.web.bind.annotation.RestController; 
 
    import com.packt.patterninspring.chapter11.
      reactivewebapp.model.Account; 
    import  com.packt.patterninspring.chapter11.
      reactivewebapp.repository.AccountRepository; 
 
    import reactor.core.publisher.Flux; 
    import reactor.core.publisher.Mono; 
 
    @RestController 
    public class AccountController { 
    
      @Autowired 
      private AccountRepository repository; 
  
      @GetMapping(value = "/account") 
      public Flux<Account> findAll() { 
        return repository.findAll().map(a -> new 
          Account(a.getId(), a.getName(),
           a.getBalance(), a.getBranch())); 
      } 
  
      @GetMapping(value = "/account/{id}") 
      public Mono<Account> findById(@PathVariable("id") Long id) { 
        return repository.findById(id) 
         .map(a -> new Account(a.getId(), a.getName(), a.getBalance(),
            a.getBranch())); 
      } 
  
      @PostMapping("/account") 
      public Mono<Account> create(@RequestBody 
        Publisher<Account> accountStream) { 
        return repository 
          .save(Mono.from(accountStream) 
          .map(a -> new Account(a.getId(), a.getName(), a.getBalance(),
             a.getBranch()))) 
          .map(a -> new Account(a.getId(), a.getName(), a.getBalance(),
             a.getBranch())); 
      } 
    } 
 

    package com.packt.patterninspring.chapter11.web.reactive.function; 
 
    import static org.springframework.http.MediaType.APPLICATION_JSON; 
    import static org.springframework.web.reactive.
      function.BodyInserters.fromObject; 
 
    import org.springframework.web.reactive.
      function.server.ServerRequest; 
    import org.springframework.web.reactive.
      function.server.ServerResponse; 
 
    import com.packt.patterninspring.chapter11.
      web.reactive.model.Account; 
    import com.packt.patterninspring.chapter11.
      web.reactive.repository.AccountRepository; 
 
    import reactor.core.publisher.Flux; 
    import reactor.core.publisher.Mono; 
 
    public class AccountHandler { 
 
      private final AccountRepository repository; 
 
      public AccountHandler(AccountRepository repository) { 
         this.repository = repository; 
      } 
 
      public Mono<ServerResponse> findById(ServerRequest request) { 
        Long accountId = Long.valueOf(request.pathVariable("id")); 
        Mono<ServerResponse> notFound = 
          ServerResponse.notFound().build(); 
        Mono<Account> accountMono =
         this.repository.findById(accountId); 
        return accountMono 
         .flatMap(account ->    ServerResponse.ok().contentType
         (APPLICATION_JSON).body(
            fromObject(account))) 
         .switchIfEmpty(notFound); 
      }  
    
      public Mono<ServerResponse> findAll(ServerRequest request) { 
       Flux<Account> accounts = this.repository.findAll(); 
       return ServerResponse.ok().contentType
       (APPLICATION_JSON).body(accounts, 
         Account.class); 
      } 
    
      public Mono<ServerResponse> create(ServerRequest request) { 
        Mono<Account> account = request.bodyToMono(Account.class); 
        return  ServerResponse.ok().build(this.
        repository.save(account)); 
      } 
    } 

    package com.packt.patterninspring.chapter11.
      reactivewebapp.repository; 
  
    import com.packt.patterninspring.chapter11.
      reactivewebapp.model.Account; 
 
    import reactor.core.publisher.Flux; 
    import reactor.core.publisher.Mono; 
 
    public interface AccountRepository { 
    
      Mono<Account> findById(Long id); 
     
      Flux<Account> findAll(); 
     
      Mono<Void> save(Mono<Account> account); 
    }

    package com.packt.patterninspring.chapter11.
      web.reactive.repository; 
 
    import java.util.Map; 
    import java.util.concurrent.ConcurrentHashMap; 
 
    import org.springframework.stereotype.Repository; 
 
    import com.packt.patterninspring.chapter11.web.
      reactive.model.Account; 
 
    import reactor.core.publisher.Flux; 
    import reactor.core.publisher.Mono; 
 
    @Repository 
    public class AccountRepositoryImpl implements AccountRepository { 
     
      private final Map<Long, Account> accountMap = new 
      ConcurrentHashMap<>();  
     
      public AccountRepositoryImpl() { 
        this.accountMap.put(1000l, new Account(1000l,
        "Dinesh Rajput", 50000l,
          "Sector-1")); 
        this.accountMap.put(2000l, new Account(2000l, 
        "Anamika Rajput", 60000l,
          "Sector-2")); 
        this.accountMap.put(3000l, new Account(3000l, 
        "Arnav Rajput", 70000l,
           "Sector-3")); 
        this.accountMap.put(4000l, new Account(4000l,
       "Adesh Rajput", 80000l,
           "Sector-4")); 
      } 
    
      @Override 
      public Mono<Account> findById(Long id) { 
        return Mono.justOrEmpty(this.accountMap.get(id)); 
      } 
 
      @Override 
      public Flux<Account> findAll() { 
        return Flux.fromIterable(this.accountMap.values()); 
      } 
 
      @Override 
      public Mono<Void> save(Mono<Account> account) { 
        return account.doOnNext(a -> { 
          accountMap.put(a.getId(), a); 
          System.out.format("Saved %s with id %d%n", a, a.getId()); 
        }).thenEmpty(Mono.empty()); 
         // return accountMono; 
      } 
    } 

    package com.packt.patterninspring.chapter11.web.reactive.function; 
 
    //Imports here 
 
    public class Server { 
 
      public static final String HOST = "localhost"; 
  
      public static final int TOMCAT_PORT = 8080; 
      public static final int REACTOR_PORT = 8181; 
    
      //main method here, download code for GITHUB 
    
      public RouterFunction<ServerResponse> routingFunction() { 
         AccountRepository repository = new AccountRepositoryImpl(); 
         AccountHandler handler = new AccountHandler(repository); 
 
         return nest(path("/account"), nest(accept(APPLICATION_JSON), 
           route(GET("/{id}"), handler::findById) 
           .andRoute(method(HttpMethod.GET), handler::findAll) 
           ).andRoute(POST("/").and(contentType
           (APPLICATION_JSON)), handler::create)); 
      } 
 
      public void startReactorServer() throws InterruptedException { 
         RouterFunction<ServerResponse> route = routingFunction(); 
         HttpHandler httpHandler = toHttpHandler(route); 
 
         ReactorHttpHandlerAdapter adapter = new
            ReactorHttpHandlerAdapter(httpHandler); 
         HttpServer server = HttpServer.create(HOST, REACTOR_PORT); 
         server.newHandler(adapter).block(); 
      } 
 
      public void startTomcatServer() throws LifecycleException { 
         RouterFunction<?> route = routingFunction(); 
         HttpHandler httpHandler = toHttpHandler(route); 
 
         Tomcat tomcatServer = new Tomcat(); 
         tomcatServer.setHostname(HOST); 
         tomcatServer.setPort(TOMCAT_PORT); 
         Context rootContext = tomcatServer.addContext("",
           System.getProperty("java.io.tmpdir")); 
         ServletHttpHandlerAdapter servlet = new
           ServletHttpHandlerAdapter(httpHandler); 
         Tomcat.addServlet(rootContext, "httpHandlerServlet", servlet); 
         rootContext.addServletMapping("/", "httpHandlerServlet"); 
         tomcatServer.start(); 
      } 
    } 

    package com.packt.patterninspring.chapter11.web.reactive.function; 
 
    //Imports here 
 
    public class Client { 
 
      private ExchangeFunction exchange = ExchangeFunctions.create(new
        ReactorClientHttpConnector()); 
 
      public void findAllAccounts() { 
        URI uri = URI.create(String.format("http://%s:%d/account",
        Server.HOST,
          Server.TOMCAT_PORT)); 
        ClientRequest request = ClientRequest.method(HttpMethod.GET,  
        uri).build(); 
 
        Flux<Account> account = exchange.exchange(request) 
        .flatMapMany(response -> response.bodyToFlux(Account.class)); 
 
         Mono<List<Account>> accountList = account.collectList(); 
         System.out.println(accountList.block()); 
      } 
 
      public void createAccount() { 
        URI uri = URI.create(String.format("http://%s:%d/account",
        Server.HOST,
           Server.TOMCAT_PORT)); 
        Account jack = new Account(5000l, "Arnav Rajput", 500000l,
        "Sector-5"); 
 
        ClientRequest request = ClientRequest.method(HttpMethod.POST,
        uri) 
        .body(BodyInserters.fromObject(jack)).build(); 
 
        Mono<ClientResponse> response = exchange.exchange(request); 
 
        System.out.println(response.block().statusCode()); 
      } 
    }