import com.amazon.sqs.javamessaging.SQSConnectionFactory; //导入依赖的package包/类
/**
 * Gets a JMS connection factory.
 *
 * @return the JMS connection factory.
 */
@Bean
public ConnectionFactory jmsConnectionFactory()
{
    AwsParamsDto awsParamsDto = awsHelper.getAwsParamsDto();

    ClientConfiguration clientConfiguration = new ClientConfiguration();

    // Only set the proxy hostname and/or port if they're configured.
    if (StringUtils.isNotBlank(awsParamsDto.getHttpProxyHost()))
    {
        clientConfiguration.setProxyHost(awsParamsDto.getHttpProxyHost());
    }
    if (awsParamsDto.getHttpProxyPort() != null)
    {
        clientConfiguration.setProxyPort(awsParamsDto.getHttpProxyPort());
    }

    return SQSConnectionFactory.builder().withClientConfiguration(clientConfiguration).build();
}
 

import com.amazon.sqs.javamessaging.SQSConnectionFactory; //导入依赖的package包/类
@Bean
public SQSConnection sqsConnection(final AWSCredentialsProvider awsCredentialsProvider,
    final ClientConfiguration awsClientConfig, final Region awsRegion) throws JMSException {
    
    SQSConnectionFactory connectionFactory = SQSConnectionFactory.builder()
        .withRegion(awsRegion) //Gets region form meta data
        .withAWSCredentialsProvider(awsCredentialsProvider)
        .withClientConfiguration(awsClientConfig)
        .build();

    return connectionFactory.createConnection();
}
 

import com.amazon.sqs.javamessaging.SQSConnectionFactory; //导入依赖的package包/类
@Bean
public SQSConnection sqsConnection(AWSCredentialsProvider awsCredentialsProvider,
                                   ClientConfiguration awsClientConfig) throws JMSException {

    SQSConnectionFactory connectionFactory = SQSConnectionFactory.builder()
            .withRegion(RegionUtils.getRegion(awsRegion))
            .withAWSCredentialsProvider(awsCredentialsProvider)
            .withClientConfiguration(awsClientConfig)
            .build();

    return connectionFactory.createConnection();
}
 

import com.amazon.sqs.javamessaging.SQSConnectionFactory; //导入依赖的package包/类
@Bean
public SQSConnectionFactory sqsConnectionFactory() throws JMSException {

    return new SQSConnectionFactory(
            new ProviderConfiguration(),
            AmazonSQSClientBuilder.standard()
                    .withRegion(Regions.AP_SOUTHEAST_2)
                    .withCredentials(new ProfileCredentialsProvider(CREDENTIALS_PROVIDER))
    );
}
 

import com.amazon.sqs.javamessaging.SQSConnectionFactory; //导入依赖的package包/类
@Override
public Connection createConnection() throws JMSException {
	Builder builder = SQSConnectionFactory.builder().withAWSCredentialsProvider(new StaticCredentialsProvider(new BasicAWSCredentials(this.accessKey,this.secretKey)));
	
	if (this.regionName != null && ! this.regionName.isEmpty()) {
		builder = builder.withRegionName(regionName);
	}
	
	this.factory = builder.build();
	
	return factory.createConnection();
}
 

import com.amazon.sqs.javamessaging.SQSConnectionFactory; //导入依赖的package包/类
public void init(AWSKafkaConfig config){
	
	this.config = config;
	deDuplicationId  = config.getDeDeupPrefix();
	
	mXmitDisable = config.getAwsXmitDisable();
	
	SQSConnectionFactory connectionFactory = new SQSConnectionFactory(
	        new ProviderConfiguration(),
	        AmazonSQSClientBuilder.standard()
	                .withRegion(config.getRegions())
	                .withCredentials(config.getCredentialsProvider())
	        );
	try{ 
		
		mPreProcessor = config.getPre();
		mXmitReplacement = config.getReplaceXmit(this);
		// Create the connection.
		connection = connectionFactory.createConnection();
		
		// Create the queue if needed
        //ExampleCommon.ensureQueueExists(connection, config.getQueueName());
            
        // Create the session
        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        producer = session.createProducer( session.createQueue( config.getQueueName() ) );
        
     // Get the wrapped client
        AmazonSQSMessagingClientWrapper client = connection.getWrappedAmazonSQSClient();

        // Create an Amazon SQS queue if it does not already exist
        if (!client.queueExists(config.getQueueName())) {
            Map<String, String> attributes = new HashMap<String, String>();
            attributes.put(config.getQueueName(), "true");
            attributes.put("ContentBasedDeduplication", "true");
            client.createQueue(new CreateQueueRequest().withQueueName(config.getQueueName()).withAttributes(attributes));
        }
	}catch(JMSException e){
	}
       
}
 

import com.amazon.sqs.javamessaging.SQSConnectionFactory; //导入依赖的package包/类
@Override
protected void starting(Description description)
{
  final String methodName = description.getMethodName();
  final String className = description.getClassName();

  testBase = new SQSTestBase();
  if (testBase.validateTestCreds() == false) {
    return;
  }
  testBase.generateCurrentQueueName(methodName);
  try {
    testBase.beforTest();
  } catch (AssumptionViolatedException ave) {
    throw ave;
  } catch (Exception e) {
    throw new RuntimeException(e);
  }

  baseDir = "target/" + className + "/" + methodName;

  Attribute.AttributeMap attributeMap = new Attribute.AttributeMap.DefaultAttributeMap();
  attributeMap.put(Context.OperatorContext.SPIN_MILLIS, 500);
  attributeMap.put(Context.DAGContext.APPLICATION_PATH, baseDir);

  context = mockOperatorContext(1, attributeMap);
  operator = new JMSStringInputOperator();
  operator.setConnectionFactoryBuilder(new JMSBase.ConnectionFactoryBuilder()
  {

    @Override
    public ConnectionFactory buildConnectionFactory()
    {
      // Create the connection factory using the environment variable credential provider.
      // Connections this factory creates can talk to the queues in us-east-1 region.
      SQSConnectionFactory connectionFactory =
          SQSConnectionFactory.builder()
          .withRegion(Region.getRegion(Regions.US_EAST_1))
          .withAWSCredentialsProvider(new PropertiesFileCredentialsProvider(testBase.getDevCredsFilePath()))
          .build();
      return connectionFactory;
    }

    @Override
    public String toString()
    {
      return className + "/" + methodName + "/ConnectionFactoryBuilder";
    }

  });
  operator.setSubject(testBase.getCurrentQueueName());
  // for SQS ack mode should be "AUTO_ACKNOWLEDGE" and transacted = false
  operator.setAckMode("AUTO_ACKNOWLEDGE");
  operator.setTransacted(false);

  sink = new CollectorTestSink<>();
  operator.output.setSink(sink);
  operator.setup(context);
  operator.activate(context);
}