<commit_before>/*
 * Copyright (c) Microsoft. All rights reserved.
 * Licensed under the MIT license. See LICENSE file in the project root for full license information.
 */
package com.microsoft.azure.eventhubs;

import java.nio.*;
import java.time.*;
import java.util.*;

import org.apache.qpid.proton.Proton;
import org.apache.qpid.proton.amqp.*;
import org.apache.qpid.proton.amqp.messaging.*;
import org.apache.qpid.proton.message.*;
import com.microsoft.azure.servicebus.amqp.*;

/**
 * The data structure encapsulating the Event being sent-to and received-from EventHubs.
 * Each EventHubs partition can be visualized as a Stream of {@link EventData}.
 */
public class EventData
{
	private String partitionKey;
	private String offset;
	private long sequenceNumber;
	private Instant enqueuedTime;
	private Binary bodyData;
	private boolean isReceivedEvent;
	private Map<String, String> properties;
	
	private SystemProperties systemProperties;
	
	private EventData()
	{
	}
	
	/**
	 * Internal Constructor - intended to be used only by the {@link PartitionReceiver} to Create #EventData out of #Message
	 */
	@SuppressWarnings("unchecked")
	EventData(Message amqpMessage)
	{
		if (amqpMessage == null)
		{
			throw new IllegalArgumentException("amqpMessage cannot be null");
		}
		
		Map<Symbol, Object> messageAnnotations = amqpMessage.getMessageAnnotations().getValue();
		
		Object partitionKeyObj = messageAnnotations.get(AmqpConstants.PARTITION_KEY);
		if (partitionKeyObj != null)
		{
			this.partitionKey = partitionKeyObj.toString();
		}
		
		Object sequenceNumberObj = messageAnnotations.get(AmqpConstants.SEQUENCE_NUMBER);
		this.sequenceNumber = (Long) sequenceNumberObj;
		
		Object enqueuedTimeUtcObj = messageAnnotations.get(AmqpConstants.ENQUEUED_TIME_UTC);
		this.enqueuedTime = ((Date) enqueuedTimeUtcObj).toInstant();
		
		this.offset = messageAnnotations.get(AmqpConstants.OFFSET).toString();
		
		this.properties = amqpMessage.getApplicationProperties() == null ? null 
				: ((Map<String, String>)(amqpMessage.getApplicationProperties().getValue()));
		
		
		this.bodyData = amqpMessage.getBody() == null ? null : ((Data) amqpMessage.getBody()).getValue();
		
		this.isReceivedEvent = true;

		amqpMessage.clear();
	}
	
	/**
	 * Construct EventData to Send to EventHubs.
	 * Typical pattern to create a Sending EventData is:<commit_msg>EventData should retain MessageAnnotations from AmqpMessage<commit_after>/*
 * Copyright (c) Microsoft. All rights reserved.
 * Licensed under the MIT license. See LICENSE file in the project root for full license information.
 */
package com.microsoft.azure.eventhubs;

import java.nio.*;
import java.time.*;
import java.util.*;

import org.apache.qpid.proton.Proton;
import org.apache.qpid.proton.amqp.*;
import org.apache.qpid.proton.amqp.messaging.*;
import org.apache.qpid.proton.message.*;
import com.microsoft.azure.servicebus.amqp.*;

/**
 * The data structure encapsulating the Event being sent-to and received-from EventHubs.
 * Each EventHubs partition can be visualized as a Stream of {@link EventData}.
 */
public class EventData
{
	private String partitionKey;
	private String offset;
	private long sequenceNumber;
	private Instant enqueuedTime;
	private Binary bodyData;
	private boolean isReceivedEvent;
	private Map<String, String> properties;
	
	private SystemProperties systemProperties;
	
	private EventData()
	{
	}
	
	/**
	 * Internal Constructor - intended to be used only by the {@link PartitionReceiver} to Create #EventData out of #Message
	 */
	@SuppressWarnings("unchecked")
	EventData(Message amqpMessage)
	{
		if (amqpMessage == null)
		{
			throw new IllegalArgumentException("amqpMessage cannot be null");
		}
		
		Map<Symbol, Object> messageAnnotations = amqpMessage.getMessageAnnotations().getValue();
		
		Object partitionKeyObj = messageAnnotations.get(AmqpConstants.PARTITION_KEY);
		if (partitionKeyObj != null)
		{
			this.partitionKey = partitionKeyObj.toString();
			messageAnnotations.remove(AmqpConstants.PARTITION_KEY);
		}
		
		Object sequenceNumberObj = messageAnnotations.get(AmqpConstants.SEQUENCE_NUMBER);
		this.sequenceNumber = (Long) sequenceNumberObj;
		messageAnnotations.remove(AmqpConstants.SEQUENCE_NUMBER);
		
		Object enqueuedTimeUtcObj = messageAnnotations.get(AmqpConstants.ENQUEUED_TIME_UTC);
		this.enqueuedTime = ((Date) enqueuedTimeUtcObj).toInstant();
		messageAnnotations.remove(AmqpConstants.ENQUEUED_TIME_UTC);
		
		this.offset = messageAnnotations.get(AmqpConstants.OFFSET).toString();
		messageAnnotations.remove(AmqpConstants.OFFSET);
		
		this.properties = amqpMessage.getApplicationProperties() == null ? null 
				: ((Map<String, String>)(amqpMessage.getApplicationProperties().getValue()));
		
		if (!messageAnnotations.isEmpty())
		{
			if (this.properties == null)
			{
				this.properties = new HashMap<String, String>();
			}
			
			for(Map.Entry<Symbol, Object> annotation: messageAnnotations.entrySet())
			{
				this.properties.put(annotation.getKey().toString(), annotation.getValue() != null ? annotation.getValue().toString() : null);
			}
		}

		this.bodyData = amqpMessage.getBody() == null ? null : ((Data) amqpMessage.getBody()).getValue();
		
		this.isReceivedEvent = true;

		amqpMessage.clear();
	}
	
	/**
	 * Construct EventData to Send to EventHubs.
	 * Typical pattern to create a Sending EventData is:
{'fraction_non_alphanumeric': 0.10899589399029488, 'fraction_numerical': 0.0, 'mean_word_length': 4.787257019438445, 'pattern_counts': {'":': 0, '<': 11, '<?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': '6963026', 'n_tokens_mistral': 1689, 'n_tokens_neox': 1522, 'n_words': 384}