import zipkin.storage.StorageComponent; //导入依赖的package包/类
@Bean
EventHubCollector eventHubCollector(EventHubCollectorProperties properties,
                                    CollectorSampler sampler,
                                    CollectorMetrics metrics,
                                    StorageComponent storage) {


  logger.log(Level.INFO,"===========EventHubCollectorAutoConfiguration==============");

  return properties.toBuilder()
      .sampler(sampler)
      .storage(storage)
      .metrics(metrics)
      .build()
      .start();
}
 

import zipkin.storage.StorageComponent; //导入依赖的package包/类
@Test
public void acceptsTrace() {
  storage = new InMemoryStorage();
  StorageConsumer storageConsumer = new StorageConsumer() {
    @Override Logger log() {
      return logger;
    }

    @Override protected StorageComponent tryCompute() {
      return storage;
    }
  };

  storageConsumer.accept(TestObjects.TRACE);
  assertThat(logger.lines())
      .extracting("level", "text")
      .containsExactly(tuple(LogLevel.DebugLevel, "Wrote 3 spans"));

  assertThat(storage.spanStore().getRawTrace(
      TestObjects.TRACE.get(0).traceIdHigh,
      TestObjects.TRACE.get(0).traceId
  )).isEqualTo(TestObjects.TRACE);
}
 

import zipkin.storage.StorageComponent; //导入依赖的package包/类
@Test(timeout = 1000L)
public void get_memoizes() throws InterruptedException {
  AtomicInteger provisionCount = new AtomicInteger();

  StorageConsumer storageConsumer = new StorageConsumer() {
    @Override protected StorageComponent tryCompute() {
      provisionCount.incrementAndGet();
      return new InMemoryStorage();
    }
  };

  int getCount = 1000;
  CountDownLatch latch = new CountDownLatch(getCount);
  ExecutorService exec = Executors.newFixedThreadPool(10);
  for (int i = 0; i < getCount; i++) {
    exec.execute(() -> {
      storageConsumer.get();
      latch.countDown();
    });
  }
  latch.await();
  exec.shutdown();
  exec.awaitTermination(1, TimeUnit.SECONDS);

  assertThat(provisionCount.get()).isEqualTo(1);
}
 

import zipkin.storage.StorageComponent; //导入依赖的package包/类
@ConditionalOnBean(StorageComponent.class)
@Bean StorageConsumer storageConsumer(
    StorageComponent component,
    @Value("${zipkin.sparkstreaming.consumer.storage.fail-fast:true}") boolean failFast,
    BeanFactory bf
) throws IOException {
  if (failFast) checkStorageOk(component);
  Properties properties = extractZipkinProperties(bf.getBean(ConfigurableEnvironment.class));
  if (component instanceof V2StorageComponent) {
    zipkin2.storage.StorageComponent v2Storage = ((V2StorageComponent) component).delegate();
    if (v2Storage instanceof ElasticsearchHttpStorage) {
      return new ElasticsearchStorageConsumer(properties);
    } else if (v2Storage instanceof zipkin2.storage.cassandra.CassandraStorage) {
      return new Cassandra3StorageConsumer(properties);
    } else {
      throw new UnsupportedOperationException(v2Storage + " not yet supported");
    }
  } else if (component instanceof CassandraStorage) {
    return new CassandraStorageConsumer(properties);
  } else if (component instanceof MySQLStorage) {
    return new MySQLStorageConsumer(properties);
  } else {
    throw new UnsupportedOperationException(component + " not yet supported");
  }
}
 

import zipkin.storage.StorageComponent; //导入依赖的package包/类
ScribeSpanConsumer newScribeSpanConsumer(String category, AsyncSpanConsumer consumer) {
  return new ScribeSpanConsumer(ScribeCollector.builder()
      .category(category)
      .metrics(scribeMetrics)
      .storage(new StorageComponent() {
        @Override public SpanStore spanStore() {
          throw new AssertionError();
        }

        @Override public AsyncSpanStore asyncSpanStore() {
          throw new AssertionError();
        }

        @Override public AsyncSpanConsumer asyncSpanConsumer() {
          return consumer;
        }

        @Override public CheckResult check() {
          return CheckResult.OK;
        }

        @Override public void close() {
          throw new AssertionError();
        }
      }));
}
 

import zipkin.storage.StorageComponent; //导入依赖的package包/类
KafkaCollector newKafkaTransport(Builder builder, AsyncSpanConsumer consumer) {
  return new KafkaCollector(builder.storage(new StorageComponent() {
    @Override public SpanStore spanStore() {
      throw new AssertionError();
    }

    @Override public AsyncSpanStore asyncSpanStore() {
      throw new AssertionError();
    }

    @Override public AsyncSpanConsumer asyncSpanConsumer() {
      return consumer;
    }

    @Override public CheckResult check() {
      return CheckResult.OK;
    }

    @Override public void close() {
      throw new AssertionError();
    }
  })).start();
}
 

import zipkin.storage.StorageComponent; //导入依赖的package包/类
@Bean
EventHubCollector eventHubCollector(ZipkinEventHubCollectorProperties properties,
    CollectorSampler sampler,
    CollectorMetrics metrics,
    StorageComponent storage) {

  return properties.toBuilder()
      .sampler(sampler)
      .storage(storage)
      .metrics(metrics)
      .build()
      .start();
}
 

import zipkin.storage.StorageComponent; //导入依赖的package包/类
final StorageComponent get() {
  StorageComponent result = instance;
  if (result == null) {
    synchronized (this) {
      result = instance;
      if (result == null) {
        instance = result = tryCompute();
      }
    }
  }
  return result;
}
 

import zipkin.storage.StorageComponent; //导入依赖的package包/类
@Override protected StorageComponent tryCompute() {
  AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
  PropertiesPropertySource source = new PropertiesPropertySource("consumer", properties);
  context.getEnvironment().getPropertySources().addLast(source);

  context.register(PropertyPlaceholderAutoConfiguration.class);
  registerAutoConfiguration(context);
  context.refresh();

  return context.getBean(StorageComponent.class);
}
 

import zipkin.storage.StorageComponent; //导入依赖的package包/类
@Override
public Object postProcessAfterInitialization(Object bean, String beanName) {
  if (bean instanceof StorageComponent && brave != null) {
    return new TracedStorageComponent(brave, (StorageComponent) bean);
  }
  return bean;
}
 

import zipkin.storage.StorageComponent; //导入依赖的package包/类
@Bean
SQSCollector sqsCollector(ZipkinSQSCollectorProperties properties, AWSCredentialsProvider credentialsProvider,
    CollectorSampler sampler, CollectorMetrics metrics, StorageComponent storage) {
  return properties.toBuilder()
      .queueUrl(properties.getQueueUrl())
      .waitTimeSeconds(properties.getWaitTimeSeconds())
      .parallelism(properties.getParallelism())
      .endpointConfiguration(endpointConfiguration)
      .credentialsProvider(credentialsProvider)
      .sampler(sampler)
      .metrics(metrics)
      .storage(storage)
      .build()
      .start();
}
 

import zipkin.storage.StorageComponent; //导入依赖的package包/类
@Bean
KinesisCollector kinesisCollector(ZipkinKinesisCollectorProperties properties,
    AWSCredentialsProvider credentialsProvider,
    CollectorSampler sampler, CollectorMetrics metrics, StorageComponent storage) {
  return KinesisCollector.builder()
      .credentialsProvider(credentialsProvider)
      .sampler(sampler)
      .metrics(metrics)
      .storage(storage)
      .streamName(properties.getStreamName())
      .appName(properties.getAppName())
      .build()
      .start();
}
 

import zipkin.storage.StorageComponent; //导入依赖的package包/类
/** lazy so transient storage errors don't crash bootstrap */
@Lazy
@Autowired
ZipkinMessageListener(StorageComponent storage, CollectorSampler sampler,
		CollectorMetrics metrics) {
	this.collector = Collector.builder(getClass())
			.storage(storage)
			.sampler(sampler)
			.metrics(metrics.forTransport("stream")).build();
}
 

import zipkin.storage.StorageComponent; //导入依赖的package包/类
@Bean
StorageComponent storage() {
    return new InMemoryStorage();
}
 

import zipkin.storage.StorageComponent; //导入依赖的package包/类
@Bean
StorageComponent storage() {
  return new InMemoryStorage();
}
 

import zipkin.storage.StorageComponent; //导入依赖的package包/类
/** Subclasses should initialize this from serializable state. */
protected abstract StorageComponent tryCompute();
 

import zipkin.storage.StorageComponent; //导入依赖的package包/类
@Override protected StorageComponent tryCompute() {
  return storage;
}
 

import zipkin.storage.StorageComponent; //导入依赖的package包/类
void checkStorageOk(StorageComponent component) throws IOException {
  Component.CheckResult result = component.check();
  if (!result.ok) throw new IllegalStateException("Storage not ok", result.exception);
  component.close(); // we don't use this directly as job instantiates their own
}
 

import zipkin.storage.StorageComponent; //导入依赖的package包/类
@Bean StorageComponent storage() {
  return new InMemoryStorage();
}
 

import zipkin.storage.StorageComponent; //导入依赖的package包/类
@Autowired ZipkinHttpCollector(StorageComponent storage, CollectorSampler sampler,
    CollectorMetrics metrics) {
  this.metrics = metrics.forTransport("http");
  this.collector = Collector.builder(getClass())
      .storage(storage).sampler(sampler).metrics(this.metrics).build();
}