<commit_before>package reactor.spring.concurrent

import org.springframework.util.concurrent.FailureCallback
import org.springframework.util.concurrent.SuccessCallback
import reactor.rx.broadcast.Broadcaster
import reactor.spring.core.concurrent.AdaptingListenableFutureProcessor
import reactor.spring.core.concurrent.ListenableFutureProcessor
import spock.lang.Specification

/**
 * Created by jbrisbin on 3/30/15.
 */
class ListenableFutureProcessorSpec extends Specification {

  def "ListenableFutureProcessor listens for values"() {

    given:
      def val = ""
      def f = new ListenableFutureProcessor()
      f.addCallback({ v -> val = v } as SuccessCallback, null)
      def b = Broadcaster.create()
      b.subscribe(f)

    when: 'a value is pushed downstream'
      b.onNext("Hello World!")

    then: 'the value is available'
      f.get() == 'Hello World!'

  }

  def "AdaptingListenableFutureProcessor adapts values"() {

    given:
      int val = 0
      def f = new AdaptingListenableFutureProcessor<String, Integer>() {
        @Override
        protected Integer adapt(String s) {
          return Integer.parseInt(s)
        }
      }
      f.addCallback({ v -> val = v } as SuccessCallback, null)
      def b = Broadcaster.<String> create()
      b.subscribe(f)

    when: 'a value is pushed downstream'
      b.onNext("1")

    then: 'the value is available as an int'
      f.get() == 1<commit_msg>Added a test for using the Subscriber API to complete the Promise.<commit_after>package reactor.spring.concurrent

import org.springframework.util.concurrent.FailureCallback
import org.springframework.util.concurrent.SuccessCallback
import reactor.rx.broadcast.Broadcaster
import reactor.spring.core.concurrent.AdaptingListenableFutureProcessor
import reactor.spring.core.concurrent.ListenableFutureProcessor
import spock.lang.Specification

/**
 * Created by jbrisbin on 3/30/15.
 */
class ListenableFutureProcessorSpec extends Specification {

  def "ListenableFutureProcessor listens for values"() {

    given:
      def val = ""
      def f = new ListenableFutureProcessor()
      f.addCallback({ v -> val = v } as SuccessCallback, null)
      def b = Broadcaster.create()
      b.subscribe(f)

    when: 'a value is pushed downstream'
      b.onNext("Hello World!")

    then: 'the value is available'
      f.get() == 'Hello World!'

  }

  def "ListenableFutureProcessor is completable"() {

    given:
      def val = ""
      def f = new ListenableFutureProcessor()
      def b = Broadcaster.create()
      b.consume {
        val = it
      }
      f.subscribe(b)

    when: 'a value is pushed downstream'
      f.onNext("Hello World!")

    then: 'the value is available'
      val == 'Hello World!'

  }

  def "AdaptingListenableFutureProcessor adapts values"() {

    given:
      int val = 0
      def f = new AdaptingListenableFutureProcessor<String, Integer>() {
        @Override
        protected Integer adapt(String s) {
          return Integer.parseInt(s)
        }
      }
      f.addCallback({ v -> val = v } as SuccessCallback, null)
      def b = Broadcaster.<String> create()
      b.subscribe(f)

    when: 'a value is pushed downstream'
      b.onNext("1")

    then: 'the value is available as an int'
      f.get() == 1
{'fraction_non_alphanumeric': 0.09942055504727051, 'fraction_numerical': 0.004879536444037816, 'mean_word_length': 2.970944309927361, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4446843', 'n_tokens_mistral': 967, 'n_tokens_neox': 911, 'n_words': 266}