class AWS::SNS::Topic

def ==(other)
  other.kind_of?(Topic) and other.arn == arn
end

def confirm_subscription(token, opts = {})
  confirm_opts = opts.merge(:token => token, :topic_arn => arn)
  resp = client.confirm_subscription(confirm_opts)
  Subscription.new(
    resp.subscription_arn,
    :topic => self,
    :config => config)
end

def delete
  client.delete_topic(:topic_arn => arn)
  nil
end

def display_name
  to_h[:display_name]
end

def display_name= display_name
  set_attribute('DisplayName', display_name)
  display_name
end

def endpoint_opts(endpoint, opts = {})
  case
  when endpoint.is_a?(SQS::Queue)
    # auto add a policy to the queue to allow the topic
    # to send the queue messages
    unless opts[:update_policy] == false
      policy = endpoint.policy || SQS::Policy.new
      policy.allow(
        :principal => :any, 
        :actions => [:send_message], 
        :resources => [endpoint]
      ).where(:source_arn).is(arn)
      endpoint.policy = policy
    end
    
    { :protocol => 'sqs', :endpoint => endpoint.arn }
  when endpoint =~ /^arn:/
    raise ArgumentError, "expected a queue ARN" unless
      endpoint =~ /^arn:aws:sqs:/
    { :protocol => "sqs", :endpoint => endpoint }
  when endpoint.kind_of?(URI)
    { :protocol => endpoint.scheme,
      :endpoint => endpoint.to_s }
  when endpoint =~ /^(https?):/
    { :protocol => $1, :endpoint => endpoint }
  when endpoint.include?("@")
    { :protocol => opts[:json] ? "email-json" : "email",
      :endpoint => endpoint }
  else
    raise ArgumentError, "could not determine protocol for '#{endpoint}'"
  end
end

def initialize arn, options = {}
  @arn = arn
  super
end

def name
  arn.split(/:/)[-1]
end

def num_subscriptions_confirmed
  to_h[:num_subscriptions_confirmed]
end

def num_subscriptions_deleted
  to_h[:num_subscriptions_deleted]
end

def num_subscriptions_pending
  to_h[:num_subscriptions_pending]
end

def owner
  to_h[:owner]
end

def parse_policy policy_json
  if policy_json
    policy = SNS::Policy.from_json(policy_json)
    policy.extend(PolicyProxy)
    policy.topic = self
    policy
  else
    nil
  end
end

def policy
  to_h[:policy]
end

def policy= policy
  policy_json = policy.is_a?(String) ? policy : policy.to_json
  set_attribute('Policy', policy_json)
  nil
end

def publish default_message, options = {}
  message = { :default => default_message }
  [:http, :https, :email, :email_json, :sqs].each do |protocol|
    if options[protocol]
      message[protocol.to_s.gsub(/_/, '-')] = options[protocol]
    end
  end
  publish_opts = {}
  publish_opts[:message] = message.to_json
  publish_opts[:message_structure] = 'json'
  publish_opts[:subject] = options[:subject] if options[:subject]
  publish_opts[:topic_arn] = arn
  
  response = client.publish(publish_opts)
  response.message_id
end

def set_attribute name, value
  client.send(:set_topic_attributes, {
    :topic_arn => arn,
    :attribute_name => name,
    :attribute_value => value,
  })
end

def subscribe(endpoint, opts = {})
  subscribe_opts = endpoint_opts(endpoint, opts).merge(:topic_arn => arn)
  resp = client.subscribe(subscribe_opts)
  if arn = resp.subscription_arn and arn =~ /^arn:/
    Subscription.new(arn, :config => config)
  else
    nil
  end
end

def subscriptions
  TopicSubscriptionCollection.new(self)
end

def to_h
  attributes = client.get_topic_attributes(:topic_arn => arn).attributes
  {
    :arn => arn,
    :name => name,
    :owner => attributes['Owner'],
    :display_name => attributes['DisplayName'] || name,
    :policy => parse_policy(attributes['Policy']),
    :num_subscriptions_confirmed => attributes['SubscriptionsConfirmed'].to_i,
    :num_subscriptions_pending => attributes['SubscriptionsPending'].to_i,
    :num_subscriptions_deleted => attributes['SubscriptionsDeleted'].to_i,
  }
end