/**
 * 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.
 */
package org.gecko.rsa.example.service;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;

import org.gecko.rsa.annotation.RequireGeckoMessageAdapterRSADiscovery;
import org.gecko.rsa.annotation.RequireGeckoMessageAdapterRSAProvider;
import org.gecko.rsa.example.api.EchoService;
import org.gecko.rsa.example.api2.EchoService2;
import org.gecko.rsa.rsatest.Address;
import org.gecko.rsa.rsatest.Contact;
import org.gecko.rsa.rsatest.ContactType;
import org.gecko.rsa.rsatest.ContextType;
import org.gecko.rsa.rsatest.Person;
import org.gecko.rsa.rsatest.RSATestFactory;
import org.osgi.service.component.annotations.Component;
import org.osgi.util.promise.Promise;
import org.osgi.util.promise.PromiseFactory;
import org.osgi.util.pushstream.PushEvent;
import org.osgi.util.pushstream.PushStream;
import org.osgi.util.pushstream.PushStreamProvider;
import org.osgi.util.pushstream.QueuePolicyOption;
import org.osgi.util.pushstream.SimplePushEventSource;

@Component(//
    property = {
                "service.exported.interfaces=*", //
                "aries.rsa.port=8201",
                "gecko.rsa.id=echo"
    })
@RequireGeckoMessageAdapterRSADiscovery
@RequireGeckoMessageAdapterRSAProvider
public class EchoServiceImpl implements EchoService , EchoService2{
	
	private final PushStreamProvider psp = new PushStreamProvider();

    @Override
    public String echo(String msg) {
        return msg + " Mark";
    }

	/* 
	 * (non-Javadoc)
	 * @see org.gecko.rsa.example.api.EchoService#getPersonById(java.lang.String)
	 */
	@Override
	public Person getPersonById(String id) {
		Person p = RSATestFactory.eINSTANCE.createPerson();
		p.setFirstName("Mark");
		p.setLastName("Hoffmann");
		p.setId(id);
		Contact c = RSATestFactory.eINSTANCE.createContact();
		c.setValue("m.hoffmann@data-in-motion.biz");
		c.setContext(ContextType.WORK);
		c.setType(ContactType.EMAIL);
		p.getContact().add(c);
		return p;
	}

	/* 
	 * (non-Javadoc)
	 * @see org.gecko.rsa.example.api.EchoService#getADdressById(java.lang.String)
	 */
	@Override
	public List<Address> getAddressById(String id) {
		List<Address> addresses = new ArrayList<Address>(2);
		Address a = RSATestFactory.eINSTANCE.createAddress();
		a.setId(id);
		a.setCity("Jena");
		a.setStreet("Kahlaische Strasse");
		a.setNumber("4");
		a.setZip("07745");
		a.setContext(ContextType.WORK);
		addresses.add(a);
		Address a2 = RSATestFactory.eINSTANCE.createAddress();
		a2.setId(id + "1");
		a2.setCity("Gera");
		a2.setStreet("Test Strasse");
		a2.setNumber("93");
		a2.setZip("07545");
		a2.setContext(ContextType.WORK);
		addresses.add(a2);
		return addresses;
	}

	/* 
	 * (non-Javadoc)
	 * @see org.gecko.rsa.example.api.EchoService#saveAddress(org.gecko.rsa.rsatest.Address)
	 */
	@Override
	public boolean saveAddress(Address a) {
		System.out.println("Saving address " + a);
		return a != null;
	}

	/* 
	 * (non-Javadoc)
	 * @see org.gecko.rsa.example.api.EchoService#getNamesByPrefix(java.lang.String)
	 */
	@Override
	public PushStream<String> getNamesByPrefix(String prefix) {
		SimplePushEventSource<String> pes = psp.buildSimpleEventSource(String.class)
				.withBuffer(new ArrayBlockingQueue<PushEvent<? extends String>>(10))
				.withQueuePolicy(QueuePolicyOption.BLOCK)
				.withExecutor(Executors.newCachedThreadPool())
				.withParallelism(2).build();
//		SimplePushEventSource<String> pes = psp.createSimpleEventSource(String.class);
		pes.connectPromise().onResolve(()-> {
			System.err.println("Connect push event source");
			for (int i = 0; i < 30; i++) {
				String s = prefix + "-Name " + (i + 1);
				pes.publish(s);
				System.err.println("Published event " + s);
				try {
					Thread.sleep(500l);
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
		});
		return psp.buildStream(pes)
				.withBuffer(new ArrayBlockingQueue<PushEvent<? extends String>>(10))
				.withQueuePolicy(QueuePolicyOption.BLOCK)
				.withExecutor(Executors.newCachedThreadPool())
				.withParallelism(2)
				.build();
	}
	
	/* 
	 * (non-Javadoc)
	 * @see org.gecko.rsa.example.api.EchoService#getData(java.lang.String)
	 */
	@Override
	public byte[] getData(String data) {
		data += ", believe me!";
		return data.getBytes();
	}

	/* 
	 * (non-Javadoc)
	 * @see org.gecko.rsa.example.api2.EchoService2#somethingElse(java.lang.String)
	 */
	@Override
	public Promise<String> somethingElse(String msg) {
		PromiseFactory pf = new PromiseFactory(Executors.newSingleThreadExecutor());
		return pf.submit(() -> msg);
	}

}