/*
 *
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 * 
 *   http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 *
 */

using System;
using System.Collections.Generic ;
using System.Threading ;
using org.apache.qpid.client ;
using org.apache.qpid.transport ;
using org.apache.qpid.transport.codec ;
using log4net ;

namespace org.apache.qpid.console
{

	/**
	 * Controls all communication with a broker. Works with the session to provide 
	 * synhchronous method calls across the asynchronous QMF bus.
	 */
	public class Broker : IMessageListener
	{
		public static ILog log = LogManager.GetLogger(typeof(Broker)) ;	
		public static int SYNC_TIME = 60000 ;
		
		public BrokerURL url ;
		public Dictionary<string, Agent> Agents = new Dictionary<string, Agent>() ;	
		
		private IClient client ;
		private IClientSession clientSession ;
		//FIXME This second session should not be needed. There is a bug in the underlieing code.
		private IClientSession outSession ;		
		private int timeout = 50000 ;
		private string replyName ;
		private string topicName ;
		private bool connected = false ;
		private bool syncInFlight = false ;		
		private bool topicBound = false ;
		private int reqsOutstanding = 0 ;
		private org.apache.qpid.console.Session consoleSession ;
		private object lockObject = new Object() ;	
		

		public Broker(org.apache.qpid.console.Session session, BrokerURL url)
		{
			log.Debug("Creating a new Broker for url " + url) ;
			this.url = url;
			consoleSession = session ;
			this.TryToConnect() ;
		}
		
		~Broker() {
			if (connected) {
				this.Shutdown() ;
			}
		}	
		
		public int BrokerBank() {
			return 1 ;
		}
		
		public bool IsConnected() {
			return connected ;
		}
		
		protected void TryToConnect() {
			reqsOutstanding = 1 ;		
			Agent newAgent = new Agent(this,0,"BrokerAgent") ;
			Agents.Add(newAgent.AgentKey(), newAgent) ;
			client = new Client() ;
			client.Connect(url.Hostname, url.Port, null, url.AuthName, url.AuthPassword) ;
			clientSession = client.CreateSession(timeout) ;		
			//clientSession.SetAutoSync(false) ;
			string name = System.Text.Encoding.UTF8.GetString(clientSession.GetName()) ;
			replyName = "reply-" + name ;
			topicName = "topic-" + name ;
			clientSession.SetAutoSync(true) ;
			Option[] options = new Option[] {Option.EXCLUSIVE, Option.AUTO_DELETE} ;
		
			// This queue is used for responses to messages which are sent.	
			clientSession.QueueDeclare(replyName,options) ;
			clientSession.ExchangeBind(replyName,"amq.direct",replyName) ;
			clientSession.AttachMessageListener(this, "rdest") ;			
			clientSession.MessageSubscribe(replyName,"rdest",MessageAcceptMode.NONE,MessageAcquireMode.PRE_ACQUIRED,null,0,null) ;			  			  						
            clientSession.MessageSetFlowMode("rdest", MessageFlowMode.WINDOW);
            clientSession.MessageFlow("rdest", MessageCreditUnit.BYTE, ClientSession.MESSAGE_FLOW_MAX_BYTES);
            clientSession.MessageFlow("rdest", MessageCreditUnit.MESSAGE, ClientSession.MESSAGE_FLOW_MAX_BYTES);  			
		
			// This queue is used for unsolicited messages sent to this class.
			clientSession.QueueDeclare(topicName, options) ;
			clientSession.AttachMessageListener(this, "tdest") ;			
			clientSession.MessageSubscribe(topicName,"tdest",MessageAcceptMode.NONE,MessageAcquireMode.PRE_ACQUIRED,null,0,null) ;							  									
            clientSession.MessageSetFlowMode("tdest", MessageFlowMode.WINDOW);
            clientSession.MessageFlow("tdest", MessageCreditUnit.BYTE, ClientSession.MESSAGE_FLOW_MAX_BYTES);
            clientSession.MessageFlow("tdest", MessageCreditUnit.MESSAGE, ClientSession.MESSAGE_FLOW_MAX_BYTES);  				
			
			outSession = client.CreateSession(timeout) ;	
			outSession.ExchangeBind(replyName,"amq.direct",replyName) ;
			
			connected = true ;
			consoleSession.HandleBrokerConnect(this) ;		
				
			
			IEncoder encoder = CreateEncoder() ;
			this.SetHeader(encoder, 'B', 0) ;
			this.Send(encoder) ;
		}
		
		public void Shutdown() {
			if (connected) {
				this.WaitForStable() ;
				clientSession.MessageStop("rdest") ;
				clientSession.MessageStop("tdest") ;
				clientSession.Close() ;
				client.Close() ;
				this.connected = false ;
			}
		}
		
		public void UpdateAgent(QMFObject obj) {
			long agentBank = (long)obj.GetProperty("agentBank") ;
			long brokerBank = (long)obj.GetProperty("brokerBank") ;
			String key = Agent.AgentKey(agentBank, brokerBank) ;
			if (obj.IsDeleted()) {
				if (Agents.ContainsKey(key)) {
					Agent agent = Agents[key] ;
					Agents.Remove(key) ;
					consoleSession.HandleAgentRemoved(agent) ;
				}
			}
			else {
				if (! Agents.ContainsKey(key)) {
					Agent newAgent = new Agent(this, agentBank, (string)obj.GetProperty("label")) ;
					Agents.Add(key, newAgent) ;
					consoleSession.HandleNewAgent(newAgent) ;
				}
			}
		}
		
		public IEncoder CreateEncoder() {
			return new MSEncoder(1000) ;	
		}
		
		
		public IEncoder CreateEncoder(char opcode, long sequence) {
			return SetHeader(this.CreateEncoder(), opcode, sequence) ;
		}
		
		public IEncoder SetHeader(IEncoder enc, char opcode, long sequence) {
			enc.WriteUint8((short)'A') ;
			enc.WriteUint8((short)'M') ;
			enc.WriteUint8((short)'2') ;
	        enc.WriteUint8((short)opcode) ;
			enc.WriteUint32(sequence) ;
			return enc ;
		}		
		
		public Message CreateMessage(IEncoder enc) {
			return this.CreateMessage(enc, "broker", -1)  ;
		}		
		
		public Message CreateMessage(IEncoder enc, string routingKey) {
			return this.CreateMessage(enc, routingKey, -1) ;
		}
		
		public Message CreateMessage(IEncoder enc, string routingKey, long ttl) {
			Message msg = new Message() ;
			msg.Body = ((MSEncoder)enc).Segment() ;
			msg.DeliveryProperties.SetRoutingKey(routingKey) ;
			if (-1 != ttl) {
				msg.DeliveryProperties.SetTtl(ttl) ;
			}
			msg.MessageProperties.SetContentType("x-application/qmf") ;
			msg.MessageProperties.SetReplyTo(new ReplyTo("amq.direct", replyName)) ;
			return msg ;
		}
		
		public void Send(IEncoder enc) {
			this.Send(this.CreateMessage(enc)) ;
		}
		
		public void Send(Message msg) {
	
			lock (lockObject) {
				log.Debug(String.Format("Sending message to routing key '{0}'", msg.DeliveryProperties.GetRoutingKey())) ;
				//log.Debug(System.Text.Encoding.UTF8.GetString(msg.Body.ToArray())) ;			
				outSession.MessageTransfer("qpid.management", msg) ;
				//clientSession.sync() ;
			}
		}
		
		protected bool CheckHeader(IDecoder decoder, out char opcode, out long sequence) {
			bool returnValue = false ;		
			opcode = 'x' ;
			sequence = -1 ;
			if(decoder.HasRemaining()) {
				char character = (char) decoder.ReadUint8() ;
				if (character != 'A') {
					return returnValue ;
				}
				character = (char) decoder.ReadUint8() ;			
				if (character != 'M') {
					return returnValue ;
				}
				character = (char) decoder.ReadUint8() ;			
				if (character != '2') {
					return returnValue ;
				}	
				returnValue = true ;
				opcode = (char) decoder.ReadUint8() ;
				sequence = decoder.ReadUint32() ;
			}
			return returnValue ;
		}
		
		public void MessageTransfer(IMessage msg) {
			MSDecoder decoder = new MSDecoder() ;
			decoder.Init(msg.Body) ;
			RangeSet rangeSet = new RangeSet() ;
			rangeSet.Add(msg.Id) ;
			char opcode = 'x' ;
			long seq = -1 ;
			while (this.CheckHeader(decoder, out opcode, out seq)) {
				//log.Debug("Message recieved with opcode " + opcode + " and sequence " + seq) ;
				//log.Debug(System.Text.Encoding.UTF8.GetString(msg.Body.ToArray())) ;
				switch (opcode) {
					case 'b':
						consoleSession.HandleBrokerResponse(this, decoder, seq) ;
						break ;
					case 'p':
						consoleSession.HandlePackageIndicator(this, decoder, seq) ;
						break ;		
					case 'z':
						consoleSession.HandleCommandComplete(this, decoder, seq) ;
						break ;		
					case 'q':
						consoleSession.HandleClassIndicator(this, decoder, seq) ;
						break ;			
					case 'm':
						consoleSession.HandleMethodResponse(this, decoder, seq) ;
						break ;							
					case 'h':
						consoleSession.HandleHeartbeatIndicator(this, decoder, seq, msg) ;
						break ;						
					case 'e':
						consoleSession.HandleEventIndicator(this, decoder, seq) ;
						break ;							
					case 's':
						consoleSession.HandleSchemaResponse(this, decoder, seq) ;
						break ;			
					case 'c':
						consoleSession.HandleContentIndicator(this, decoder, seq, true, false) ;
						break ;	
					case 'i':
						consoleSession.HandleContentIndicator(this, decoder, seq, false, true) ;
						break ;		
					case 'g':
						consoleSession.HandleContentIndicator(this, decoder, seq, true, true) ;
						break ;			
					default:
						log.Error("Invalid message type recieved with opcode " + opcode) ;
						break ;
				}		
			} 
			lock (lockObject) {
				outSession.MessageAccept(rangeSet) ;
			}
		}
		
		public void IncrementOutstanding() {
			lock (lockObject) {
				this.reqsOutstanding += 1 ;	
			}
		}
		
		public void DecrementOutstanding() {
			lock (lockObject) {
				this.reqsOutstanding -= 1 ;	
				if ((reqsOutstanding == 0) & !topicBound) {
					foreach (string key in consoleSession.BindingKeys()) {
						//this.clientSession.ExchangeBind(topicName, "qpid.mannagement", key) ;
						log.Debug("Setting Topic Binding " + key) ;						
						this.outSession.ExchangeBind(topicName, "qpid.management", key) ;
					}
					topicBound = true ;
				}
				if ((reqsOutstanding == 0) & syncInFlight) {
					syncInFlight = false ;
					Monitor.PulseAll(lockObject) ;	
				}
			}
		}	
		
		public void WaitForStable() {
			lock (lockObject) {
				if (connected) {
					DateTime start = DateTime.Now ;
					syncInFlight = true ;					
					while (reqsOutstanding != 0) {					
						log.Debug("Waiting to recieve messages") ;
						Monitor.Wait(lockObject,SYNC_TIME) ;
						TimeSpan duration = DateTime.Now - start;						
						if (duration.TotalMilliseconds > SYNC_TIME) {
							throw new Exception("Timeout waiting for Broker to Sync") ;
						}
					}
				}
			}			
		}	
		
		public void SetSyncInFlight(bool inFlight) {
			lock(lockObject) {
				syncInFlight = inFlight ;
				Monitor.PulseAll(lockObject) ;
			}
		}
		
		public bool GetSyncInFlight() {
			return syncInFlight ;
		}		
		
		public void WaitForSync(int timeout) {
			lock(lockObject) {
				DateTime start = DateTime.Now ;			
				while (syncInFlight) {
					Monitor.Wait(lockObject,timeout) ;
				}
				TimeSpan duration = DateTime.Now - start;						
				if (duration.TotalMilliseconds > timeout) {
					throw new Exception("Timeout waiting for Broker to Sync") ;
				}
			} 
		}
	}
}