Wing of Dream 梦境之翼

public class Crawler {
   private ConcurrentHashMap<String, Boolean> seen = new ConcurrentHashMap<String, Boolean>();
   private AtomicInteger pending = new AtomicInteger(0);
   
   public Crawler(String baseUrl, int numOfThreads) {
       this.client = HttpClientBuilder.create().build();
       this.baseUrl = baseUrl;
       this.executorService = Executors.newFixedThreadPool(numOfThreads, new ThreadFactory() {
          public Thread newThread(Runnable r) {
              return new Thread(r, "Crawler-Worker");
           }
       });
   }
   
   public void start() {
       handle(baseUrl);
   }
   
   private void handle(final String link) {        
      if (seen.containsKey(link))
          return;
      seen.put(link, true);
      pending.incrementAndGet();
      executorService.execute(new Runnable() {
          public void run() {
              List<String> links = getLinksFromUrl(link);
              for (String link : links) {
                  handle(link);
              }
              pending.decrementAndGet();
              if (pending.get() == 0) {
                  synchronized (lock) {
                      lock.notify();
                  }
              }
          }
      });
   }
}

public class Main{
  public static void main(String[] args) throws InterruptedException {
          if (args.length != 2) {
              System.err.println("Invalid syntax: <baseUrl> <numOfThreads>");
              System.exit(1);
          }
          String baseUrl = args[0];
          int numOfThreads = Integer.parseInt(args[1]);
          Crawler crawler = new Crawler(baseUrl, numOfThreads);
          crawler.start();
          crawler.join();
          crawler.shutdown();
  }
}

public class ObservableCrawler {
  private Subscriber<? super String> subscriber;
  
  public static Observable<String> create(Crawler crawler, String url, int numOfThreads) {
      ObservableCrawler o = new ObservableCrawler(crawler, numOfThreads);
      return Observable.create(subscriber -> {
          o.subscriber = subscriber;
          if (o.executorService == null) {
              o.process(url);
              subscriber.onCompleted();
          } else {
              o.processAsync(url);
          }
      });
  }
  private ObservableCrawler(Crawler crawler, int numOfThreads) {
      this.crawler = crawler;
      this.executorService = numOfThreads > 0 ?
              Executors.newFixedThreadPool(numOfThreads, r -> new Thread(r, "Crawler-" + threadIdGenerator.incrementAndGet()))
              : null;
  }
  
  private void processAsync(String url) {
         pendingTasks.incrementAndGet();
         executorService.submit(() -> {
             // If item is not unique, skip processing
             boolean isFirstTime =  results.add(url);
             if (isFirstTime) {
                 subscriber.onNext(url);
                 crawler.crawl(url, this::processAsync);
             }
             if (pendingTasks.decrementAndGet() == 0) {
                 subscriber.onCompleted();
                 executorService.shutdown();
             }
         });
  }
}

public class CrawlerClient {
  public CrawlerClient(Crawler crawler, String url, int numOfThreads) {
      this.observable = ObservableCrawler.create(crawler, url, numOfThreads);
      observable.subscribe(this::onNext, this::onError, this::onCompleted);
  }

  public void waitForCompletion() throws InterruptedException {
      completionLatch.await();
  }
}

-module(counter).
-export([run/0, counter/1]).

run() ->
    S = spawn(counter, counter, [0]),
    send_msgs(S, 100000),
    S.

counter(Sum) ->
    receive
        value -> io:fwrite("Value is ~w~n", [Sum]);
        {inc, Amount} -> counter(Sum+Amount)
    end.

send_msgs(_, 0) -> true;
send_msgs(S, Count) ->
    S ! {inc, 1},
    send_msgs(S, Count-1).

% Usage:
%    1> c(counter).
%    2> S = counter:run().
%       ... Wait a bit until all children have run ...
%    3> S ! value.
%    Value is 100000

object HelloWorld {
  final case class Greet(whom: String, replyTo: ActorRef[Greeted])
  final case class Greeted(whom: String, from: ActorRef[Greet])

  def apply(): Behavior[Greet] = Behaviors.receive { (context, message) =>
    context.log.info("Hello {}!", message.whom)
    message.replyTo ! Greeted(message.whom, context.self)
    Behaviors.same
  }
}

public class HelloWorld extends AbstractBehavior<HelloWorld.Greet> {

  public static final class Greet {
    public final String whom;
    public final ActorRef<Greeted> replyTo;

    public Greet(String whom, ActorRef<Greeted> replyTo) {
      this.whom = whom;
      this.replyTo = replyTo;
    }
  }

  public static final class Greeted {
    public final String whom;
    public final ActorRef<Greet> from;

    public Greeted(String whom, ActorRef<Greet> from) {
      this.whom = whom;
      this.from = from;
    }
  }

  public static Behavior<Greet> create() {
    return Behaviors.setup(HelloWorld::new);
  }

  private HelloWorld(ActorContext<Greet> context) {
    super(context);
  }

  @Override
  public Receive<Greet> createReceive() {
    return newReceiveBuilder().onMessage(Greet.class, this::onGreet).build();
  }

  private Behavior<Greet> onGreet(Greet command) {
    getContext().getLog().info("Hello {}!", command.whom);
    command.replyTo.tell(new Greeted(command.whom, getContext().getSelf()));
    return this;
  }
}